bn_user
/
sigma
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Обновить dags/OSV_with_docs.py

This commit is contained in:
bn_user 2025-11-30 18:12:15 +00:00
parent b6c516037c
commit b20a7484e5
1 changed files with 171 additions and 183 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

336
dags/OSV_with_docs.py
View File

@ -52,203 +52,191 @@ def read_data_1C(**kwargs):
# Создаем диапазон дат от начала до T-2 # Создаем диапазон дат от начала до T-2
start_date = f'{start_year}-01-01' start_date = f'{start_year}-01-01'
date_range = pd.date_range(start=start_date, end=t_minus_2, freq='D') # date_range = pd.date_range(start=start_date, end=t_minus_2, freq='D')
# Создаем пустой список для хранения данных # Создаем пустой список для хранения данных
calendar_data = [] calendar_data = []
params = {"СписокСчетов": ["66","66.01","66.02", "66.03","66.04","66.21","66.22","66.23","66.24", "67","67.01","67.02", "67.03","67.04","67.21","67.22","67.23","67.24"]} params = {"СписокСчетов": ["66","66.01","66.02", "66.03","66.04","66.21","66.22","66.23","66.24", "67","67.01","67.02", "67.03","67.04","67.21","67.22","67.23","67.24"]}
engine = get_db_engine() engine = get_db_engine()
table_name = 'oborotno_salbdovaya_vedomostb' table_name = 'oborotno_salbdovaya_vedomostb_with_docs'
temp_table_name = f'temp_{table_name}' temp_table_name = f'temp_{table_name}'
# Проходим циклом по каждому дню # Проходим циклом по каждому дню
print("\nПрохождение по дням календаря:") # print("\nПрохождение по дням календаря:")
for i, date in enumerate(date_range): # for i, date in enumerate(date_range):
print(f'{date}') # print(f'{date}')
query = f"""ВЫБРАТЬ query = f"""ВЫБРАТЬ
ОстаткиОбороты.Счет, ОстаткиОбороты.Счет,
ОстаткиОбороты.Субконто1, ОстаткиОбороты.Регистратор,
ОстаткиОбороты.Субконто2, ОстаткиОбороты.Субконто1,
ОстаткиОбороты.Организация, ОстаткиОбороты.Субконто2,
ОстаткиОбороты.Субконто2.Номер КАК НомерДоговора, ОстаткиОбороты.Организация,
ОстаткиОбороты.Субконто2.Дата КАК ДатаДоговора, ОстаткиОбороты.Субконто2.Номер КАК НомерДоговора,
ОстаткиОбороты.Субконто2.СрокДействия КАК СрокДействияДоговора, ОстаткиОбороты.Субконто2.Дата КАК ДатаДоговора,
ОстаткиОбороты.Субконто1.Инн КАК ИннКонтрагента, ОстаткиОбороты.Субконто2.СрокДействия КАК СрокДействияДоговора,
ОстаткиОбороты.Организация.Инн КАК ИннКлиента, ОстаткиОбороты.Субконто1.Инн КАК ИннКонтрагента,
ОстаткиОбороты.СуммаОборот, ОстаткиОбороты.Организация.Инн КАК ИннКлиента,
ОстаткиОбороты.СуммаОборотДт, ОстаткиОбороты.СуммаОборот,
ОстаткиОбороты.СуммаОборотКт, ОстаткиОбороты.СуммаОборотДт,
ОстаткиОбороты.СуммаКонечныйОстаток, ОстаткиОбороты.СуммаОборотКт,
ОстаткиОбороты.СуммаКонечныйОстатокДт, ОстаткиОбороты.СуммаКонечныйОстаток,
ОстаткиОбороты.СуммаКонечныйОстатокКт, ОстаткиОбороты.СуммаКонечныйОстатокДт,
ОстаткиОбороты.СуммаКонечныйРазвернутыйОстатокДт, ОстаткиОбороты.СуммаКонечныйОстатокКт,
ОстаткиОбороты.СуммаКонечныйРазвернутыйОстатокКт, ОстаткиОбороты.СуммаКонечныйРазвернутыйОстатокДт,
UUID(ОстаткиОбороты.Субконто2.Ссылка) КАК ИдентификаторДоговора, ОстаткиОбороты.СуммаКонечныйРазвернутыйОстатокКт,
UUID(ОстаткиОбороты.Субконто1.Ссылка) КАК ИдентификаторКонтрагента, UUID(ОстаткиОбороты.Регистратор.ССылка) КАК ИдентификаторРегистратора,
UUID(ОстаткиОбороты.Организация.Ссылка) КАК ИдентификаторКлиента, UUID(ОстаткиОбороты.Субконто2.Ссылка) КАК ИдентификаторДоговора,
ОстаткиОбороты.Период КАК ДатаОтчета UUID(ОстаткиОбороты.Субконто1.Ссылка) КАК ИдентификаторКонтрагента,
ИЗ UUID(ОстаткиОбороты.Организация.Ссылка) КАК ИдентификаторКлиента,
РегистрБухгалтерии.Хозрасчетный.ОстаткиИОбороты( ДАТАВРЕМЯ({date.year},{date.month},{date.day}), ДАТАВРЕМЯ({date.year},{date.month},{date.day}), Регистратор, , Счет.Код в (&СписокСчетов), , ) КАК ОстаткиОбороты ОстаткиОбороты.Период КАК ДатаОтчета
""" ИЗ
РегистрБухгалтерии.Хозрасчетный.ОстаткиИОбороты( , ДАТАВРЕМЯ({t_minus_2.year},{t_minus_2.month},{t_minus_2.day}), Регистратор, , Счет.Код в (&СписокСчетов), , ) КАК ОстаткиОбороты
"""
# РегистрБухгалтерии.Хозрасчетный.ОстаткиИОбороты( ДАТАВРЕМЯ({date.year},{date.month},{date.day}), ДАТАВРЕМЯ({date.year},{date.month},{date.day}), Регистратор, , Счет.Код в (&СписокСчетов), , ) КАК ОстаткиОбороты
auth = HTTPBasicAuth('obmen', 'bOR2W7w4')
response = requests.post(
# http://адрес сервера/название БД/hs/services/query?ProcessPostedRequest
auth = HTTPBasicAuth('obmen', 'bOR2W7w4') url=r'http://192.168.1.75/chupd/hs/services/query?ProcessPostedRequest',
response = requests.post( json={"query":query, "params": params},
# http://адрес сервера/название БД/hs/services/query?ProcessPostedRequest auth=auth,
verify=False
)
url=r'http://192.168.1.75/chupd/hs/services/query?ProcessPostedRequest', data_from_1c = response.json()
json={"query":query, "params": params}, df = pd.DataFrame(data_from_1c['data'])
auth=auth,
verify=False
)
data_from_1c = response.json() field_mapping = {
df = pd.DataFrame(data_from_1c['data']) # Основные поля
'Счет': 'schet',
'Регистратор': 'registrator',
'Субконто1': 'subkonto1',
'Субконто2': 'subkonto2',
'Организация': 'organizaciya',
'НомерДоговора': 'nomer',
'ДатаДоговора': 'date_begin',
'СрокДействияДоговора': 'date_end',
'ИннКонтрагента': 'inn_subkonto1',
'ИннКлиента': 'inn_organizaciya',
field_mapping = { # Суммовые остатки и обороты
# Основные поля 'СуммаОборот': 'summa_oborot',
'Счет': 'schet', 'СуммаОборотДт': 'summa_oborot_dt',
'Субконто1': 'subkonto1', 'СуммаОборотКт': 'summa_oborot_kt',
'Субконто2': 'subkonto2', 'СуммаКонечныйОстаток': 'summa_konechnyy_ostatok',
'Организация': 'organizaciya', 'СуммаКонечныйОстатокДт': 'summa_konechnyy_ostatok_dt',
'НомерДоговора': 'nomer', 'СуммаКонечныйОстатокКт': 'summa_konechnyy_ostatok_kt',
'ДатаДоговора': 'date_begin', 'СуммаКонечныйРазвернутыйОстатокДт': 'summa_konechnyy_razvernutyy_ostatok_dt',
'СрокДействияДоговора': 'date_end', 'СуммаКонечныйРазвернутыйОстатокКт': 'summa_konechnyy_razvernutyy_ostatok_kt',
'ИннКонтрагента': 'inn_subkonto1',
'ИннКлиента': 'inn_organizaciya',
# Суммовые остатки и обороты #ID
'СуммаОборот': 'summa_oborot', 'ИдентификаторРегистратора': 'uid_registrator',
'СуммаОборотДт': 'summa_oborot_dt', 'ИдентификаторДоговора': 'uid_subkonto2',
'СуммаОборотКт': 'summa_oborot_kt', 'ИдентификаторКонтрагента': 'uid_subkonto1',
'СуммаКонечныйОстаток': 'summa_konechnyy_ostatok', 'ИдентификаторКлиента': 'uid_organizaciya',
'СуммаКонечныйОстатокДт': 'summa_konechnyy_ostatok_dt',
'СуммаКонечныйОстатокКт': 'summa_konechnyy_ostatok_kt',
'СуммаКонечныйРазвернутыйОстатокДт': 'summa_konechnyy_razvernutyy_ostatok_dt',
'СуммаКонечныйРазвернутыйОстатокКт': 'summa_konechnyy_razvernutyy_ostatok_kt',
#ID #DATE
'ИдентификаторДоговора': 'uid_subkonto2', 'Период': 'get_date'
'ИдентификаторКонтрагента': 'uid_subkonto1', }
'ИдентификаторКлиента': 'uid_organizaciya',
#DATE df = df.rename(columns=field_mapping)
'ДатаОтчета': 'get_date' # # Удаляем дубликаты по ключевым полям перед вставкой
} # group_columns = [
# 'schet', 'subkonto1', 'subkonto2', 'organizaciya', 'nomer',
# 'date_begin', 'date_end', 'inn_subkonto1', 'inn_organizaciya',
# 'uid_subkonto2', 'uid_subkonto1', 'uid_organizaciya', 'get_date'
# ]
# columns_for_sum = [
# 'summa_oborot', 'summa_oborot_dt', 'summa_oborot_kt',
# 'summa_konechnyy_ostatok', 'summa_konechnyy_ostatok_dt', 'summa_konechnyy_ostatok_kt',
# 'summa_konechnyy_razvernutyy_ostatok_dt', 'summa_konechnyy_razvernutyy_ostatok_kt'
# ]
# sum_columns = {
# 'summa_oborot': 'sum', 'summa_oborot_dt': 'sum', 'summa_oborot_kt': 'sum',
# 'summa_konechnyy_ostatok': 'sum', 'summa_konechnyy_ostatok_dt': 'sum', 'summa_konechnyy_ostatok_kt': 'sum',
# 'summa_konechnyy_razvernutyy_ostatok_dt': 'sum', 'summa_konechnyy_razvernutyy_ostatok_kt': 'sum'
# }
# for col in columns_for_sum:
# if col in df.columns:
# # Очищаем и преобразуем числовые значения
# df[col] = (df[col]
# .str.replace('\xa0', '', regex=False) # Убираем неразрывные пробелы
# .str.replace(' ', '', regex=False) # Убираем обычные пробелы
# .str.replace(',', '.', regex=False) # Запятые -> точки для десятичных
# .apply(lambda x: x if x.replace('.', '').replace('-', '').isdigit() else None)
# .apply(pd.to_numeric, errors='coerce'))
df = df.rename(columns=field_mapping) # # Диагностика
# Удаляем дубликаты по ключевым полям перед вставкой # original = df[col].copy()
group_columns = [ # cleaned = df[col]
'schet', 'subkonto1', 'subkonto2', 'organizaciya', 'nomer', # failed = cleaned.isna() & original.notna()
'date_begin', 'date_end', 'inn_subkonto1', 'inn_organizaciya',
'uid_subkonto2', 'uid_subkonto1', 'uid_organizaciya', 'get_date'
]
columns_for_sum = [
'summa_oborot', 'summa_oborot_dt', 'summa_oborot_kt',
'summa_konechnyy_ostatok', 'summa_konechnyy_ostatok_dt', 'summa_konechnyy_ostatok_kt',
'summa_konechnyy_razvernutyy_ostatok_dt', 'summa_konechnyy_razvernutyy_ostatok_kt'
]
sum_columns = {
'summa_oborot': 'sum', 'summa_oborot_dt': 'sum', 'summa_oborot_kt': 'sum',
'summa_konechnyy_ostatok': 'sum', 'summa_konechnyy_ostatok_dt': 'sum', 'summa_konechnyy_ostatok_kt': 'sum',
'summa_konechnyy_razvernutyy_ostatok_dt': 'sum', 'summa_konechnyy_razvernutyy_ostatok_kt': 'sum'
}
for col in columns_for_sum:
if col in df.columns:
# Очищаем и преобразуем числовые значения
df[col] = (df[col]
.str.replace('\xa0', '', regex=False) # Убираем неразрывные пробелы
.str.replace(' ', '', regex=False) # Убираем обычные пробелы
.str.replace(',', '.', regex=False) # Запятые -> точки для десятичных
.apply(lambda x: x if x.replace('.', '').replace('-', '').isdigit() else None)
.apply(pd.to_numeric, errors='coerce'))
# Диагностика # if failed.any():
original = df[col].copy() # display(f" {col}: не преобразовано {failed.sum()} значений")
cleaned = df[col] # display(f" Примеры: {original[failed].head(3).tolist()}")
failed = cleaned.isna() & original.notna()
if failed.any(): # df = df.groupby(group_columns, as_index=False).agg(sum_columns)
display(f" {col}: не преобразовано {failed.sum()} значений")
display(f" Примеры: {original[failed].head(3).tolist()}")
df = df.groupby(group_columns, as_index=False).agg(sum_columns) # Преобразуем даты из формата DD.MM.YYYY HH:MM:SS в datetime
date_columns = ['date_begin', 'date_end', 'get_date']
# Преобразуем даты из формата DD.MM.YYYY HH:MM:SS в datetime for col in date_columns:
date_columns = ['date_begin', 'date_end', 'get_date'] if col in df.columns:
# Пробуем разные форматы дат
df[col] = pd.to_datetime(df[col], format='%d.%m.%Y %H:%M:%S', errors='coerce')
for col in date_columns: # Если не сработало, пробуем без времени
if col in df.columns: if df[col].isna().any():
# Пробуем разные форматы дат df[col] = pd.to_datetime(df[col], format='%d.%m.%Y', errors='coerce')
df[col] = pd.to_datetime(df[col], format='%d.%m.%Y %H:%M:%S', errors='coerce')
# Если не сработало, пробуем без времени # Убедимся, что все даты преобразованы
if df[col].isna().any(): for col in date_columns:
df[col] = pd.to_datetime(df[col], format='%d.%m.%Y', errors='coerce') if col in df.columns:
failed_count = df[col].isna().sum()
# Убедимся, что все даты преобразованы if failed_count > 0:
for col in date_columns: print(f" Внимание: в колонке '{col}' осталось {failed_count} некорректных дат")
if col in df.columns:
failed_count = df[col].isna().sum()
if failed_count > 0:
print(f" Внимание: в колонке '{col}' осталось {failed_count} некорректных дат")
with engine.begin() as conn: with engine.begin() as conn:
if not df.empty: if not df.empty:
conn.execute(f"DROP TABLE IF EXISTS {temp_table_name}") conn.execute(f"DROP TABLE IF EXISTS {temp_table_name}")
conn.execute(f"CREATE TEMP TABLE {temp_table_name} (LIKE public.{table_name})") conn.execute(f"CREATE TEMP TABLE {temp_table_name} (LIKE public.{table_name})")
df.to_sql( df.to_sql(
f'temp_{table_name}', f'temp_{table_name}',
con=conn, con=conn,
if_exists='append', if_exists='append',
index=False, index=False,
method='multi' method='multi'
) )
# conn.execute(f"DELETE FROM public.{table_name} where get_date = (SELECT DISTINCT get_date FROM temp_{table_name})") conn.execute(f"TRUNCATE TABLE public.{table_name}")
conn.execute(f""" conn.execute(f"""
INSERT INTO public.{table_name} INSERT INTO public.{table_name}
SELECT SELECT
schet schet
, subkonto1 , registrator
, subkonto2 , subkonto1
, organizaciya , subkonto2
, case when nomer = '' then null else nomer end nomer , organizaciya
, case when date_begin = '0001-01-01' then null else date_begin end date_begin , case when nomer = '' then null else nomer end nomer
, case when date_begin = '0001-01-01' then null else date_end end date_end , case when date_begin = '0001-01-01' then null else date_begin end date_begin
, inn_subkonto1 as inn_subkonto1 , case when date_begin = '0001-01-01' then null else date_end end date_end
, inn_organizaciya as inn_organizaciya , inn_subkonto1 as inn_subkonto1
, summa_oborot , inn_organizaciya as inn_organizaciya
, summa_oborot_dt , summa_oborot
, summa_oborot_kt , summa_oborot_dt
, summa_konechnyy_ostatok , summa_oborot_kt
, summa_konechnyy_ostatok_dt , summa_konechnyy_ostatok
, summa_konechnyy_ostatok_kt , summa_konechnyy_ostatok_dt
, summa_konechnyy_razvernutyy_ostatok_dt , summa_konechnyy_ostatok_kt
, summa_konechnyy_razvernutyy_ostatok_kt , summa_konechnyy_razvernutyy_ostatok_dt
, uid_subkonto2 , summa_konechnyy_razvernutyy_ostatok_kt
, uid_subkonto1 , uid_registrator
, uid_organizaciya , uid_subkonto2
, get_date::date::timestamp , uid_subkonto1
FROM temp_{table_name} , uid_organizaciya
ON CONFLICT (schet, uid_subkonto2, uid_subkonto1, uid_organizaciya, get_date) , get_date::date::timestamp
DO UPDATE SET FROM temp_{table_name}
subkonto1 = EXCLUDED.subkonto1, """
subkonto2 = EXCLUDED.subkonto2, )
organizaciya = EXCLUDED.organizaciya,
nomer = EXCLUDED.nomer,
date_begin = EXCLUDED.date_begin,
date_end = EXCLUDED.date_end,
inn_subkonto1 = EXCLUDED.inn_subkonto1,
inn_organizaciya = EXCLUDED.inn_organizaciya,
summa_oborot = EXCLUDED.summa_oborot,
summa_oborot_dt = EXCLUDED.summa_oborot_dt,
summa_oborot_kt = EXCLUDED.summa_oborot_kt,
summa_konechnyy_ostatok = EXCLUDED.summa_konechnyy_ostatok,
summa_konechnyy_ostatok_dt = EXCLUDED.summa_konechnyy_ostatok_dt,
summa_konechnyy_ostatok_kt = EXCLUDED.summa_konechnyy_ostatok_kt,
summa_konechnyy_razvernutyy_ostatok_dt = EXCLUDED.summa_konechnyy_razvernutyy_ostatok_dt,
summa_konechnyy_razvernutyy_ostatok_kt = EXCLUDED.summa_konechnyy_razvernutyy_ostatok_kt
"""
)
with DAG( with DAG(
dag_id='data_download_from_1C_source', dag_id='data_download_from_1C_source',