etl_bill_detail.py

@@ -5,7 +5,7 @@
 
 import pandas as pd
 from sqlalchemy import create_engine, text
-from sqlalchemy.types import NVARCHAR, Date, DECIMAL
+from sqlalchemy.types import NVARCHAR, Date, DECIMAL, INTEGER
 from datetime import datetime, timedelta
 from tqdm import tqdm
 import json
@@ -16,7 +16,8 @@ from utils import load_config, truncate_target_db
 
 debug = False
 if debug:
-    debug_condition = "and bd.biz_id='1593145414334246914'"
+    debug_condition = "and bd.biz_id='1687400201327599617'"  # P.53
+    # debug_condition = ''
 else:
     debug_condition = ''
 
@@ -34,15 +35,15 @@ def extract(conn, batch_size, i) -> pd.DataFrame:
     """
     query = """
         select 
-            bd.tenant_id, rc.id 'contract_id', dept.id 'dept_id', dept.name 'dept_name', rc.maintainer_id 'emp_id', emp.name 'emp_name',
+            bd.tenant_id, rc.id 'contract_id', dept.id 'dept_id', dept.name 'dept_name', rc.sign_emp_id 'emp_id', emp.name 'emp_name',
             bd.bill_id, bd.id as 'bill_detail_id', bd.fee_subject_id, sd.label as 'fee_subject_label', sd.name as 'fee_subject_name', 
             bd.fee_direction, bd.original_money, bd.occurred_money,
-            bd.begin_time, bd.end_time, bd.is_occur, rc.cancel_info
+            bd.begin_time, bd.end_time, bd.is_occur, rc.cancel_info, bd.predict_time
         from yuxin_finance.fin_finance_bill_detail bd
         left join yuxin_setting.setting_dictionary sd on sd.id=bd.fee_subject_id
         left join yuxin_contract.cont_renter_contract rc on rc.id=bd.biz_id
         left join yuxin_setting.setting_employee_dept ed on ed.emp_id=rc.maintainer_id and ed.is_delete=0
-        left join yuxin_setting.setting_employee_info emp on emp.id=rc.maintainer_id and emp.is_delete=0
+        left join yuxin_setting.setting_employee_info emp on emp.id=rc.sign_emp_id and emp.is_delete=0
         left join yuxin_setting.setting_department dept on dept.id=ed.dept_id and dept.is_delete=0
         where bd.is_valid=1 and bd.is_delete=0 and bd.biz_type=2 {debug_condition}
         limit {batch_size} offset {offset}
@@ -63,7 +64,8 @@ def transform(data) -> pd.DataFrame:
                     'reject_name',
                     'reject_payment_account',
                     'reject_payment_account_type',
-                    'reject_reason'
+                    'reject_reason',
+                    'is_apportion',
                     ])
     # target data
     df = pd.DataFrame(columns=columns)
@@ -112,6 +114,7 @@ def transform(data) -> pd.DataFrame:
             elif reject_payment_account_type == 3:
                 reject_payment_account_type = '微信'
         
+        # 权责发生制
         # Loop from begin_date to end_date by day
         remainder = row['original_money']
         current_month = begin_date
@@ -127,6 +130,7 @@ def transform(data) -> pd.DataFrame:
             # Copy the row and add extra columns
             new_row = row.copy()
             new_row['day'] = current_month.date()
+            new_row['is_apportion'] = 1
             
             if current_month.month == end_date.month:
                 # keep remainder in the last day
@@ -170,6 +174,32 @@ def transform(data) -> pd.DataFrame:
             # current_month += timedelta(days=1)
             current_month += relativedelta(months=1)
 
+        # 收付实现制
+        # 定金、押金
+        if row['fee_subject_label'] == 'FEESUBJECT@DEPOSIT':
+            # print('收付实现制', row['fee_subject_label'])
+            new_row = row.copy()
+            new_row['day'] = row['predict_time']
+            new_row['is_apportion'] = 0
+            if row['fee_direction'] == 1:
+                if row['is_occur'] == 1:
+                    new_row['kind'] = '实收'
+                else:
+                    new_row['kind'] = '应收'
+            else:
+                if row['is_occur'] == 1:
+                    new_row['kind'] = '实付'
+                else:
+                    new_row['kind'] = '应付'
+            new_row['money'] = row['original_money']
+            new_row['reject_time'] = reject_time
+            new_row['reject_name'] = reject_name
+            new_row['reject_payment_account'] = reject_payment_account if reject_payment_account is not None else ''
+            new_row['reject_payment_account_type'] = reject_payment_account_type if reject_payment_account_type is not None else ''
+            new_row['reject_reason'] = reject_reason if reject_reason is not None else ''
+            df.loc[len(df)] = new_row
+
+
     return df
 
 
@@ -198,6 +228,7 @@ def load(conn, df: pd.DataFrame, target_db) -> None:
         'reject_payment_account': NVARCHAR(64),
         'reject_payment_account_type': NVARCHAR(32),
         'reject_reason': NVARCHAR(64),
+        'is_apportion': INTEGER,
     }
     # create target table with df.dtypes
     df.to_sql(target_db, con=conn, if_exists='append',
@@ -207,7 +238,7 @@ def load(conn, df: pd.DataFrame, target_db) -> None:
 
 
 def etl():
-    config = load_config('production')
+    config = load_config()
 
     target_db = 'bi_bill_detail_'
 
@@ -229,9 +260,8 @@ def etl():
         truncate_target_db(conn, target_db)
 
         if debug:
-            total = 1
-            batch_size = 1
-        # total = 200
+            total = 200
+            batch_size = 100
         print('total', total)
 
         # Write the data to the table in batches