import os import re import json import io import zipfile import pandas as pd import xml.etree.ElementTree as ET from datetime import datetime import xml.dom.minidom import streamlit as st MAPPING_FILE = os.path.join(os.path.dirname(__file__), "term_mapping.json") def clean_id(val): if pd.isna(val): return "" s = str(val).strip() if '.' in s and s.split('.')[1].isdigit() and int(s.split('.')[1]) == 0: return s.split('.')[0] if 'E' in s.upper(): try: return str(int(float(s))) except: pass return s def parse_decimal(val): if pd.isna(val): return 0.0 if isinstance(val, (int, float)): return float(val) s = str(val).replace('\xa0', '').replace(' ', '').replace(',', '.') try: return float(s) except: return 0.0 def parse_date(val): if pd.isna(val): return "" if isinstance(val, (datetime, pd.Timestamp)): return val.strftime("%Y-%m-%d") try: dt = datetime.strptime(str(val).strip()[:10], "%d.%m.%Y") return dt.strftime("%Y-%m-%d") except: return str(val)[:10] def parse_datetime(val): if pd.isna(val): return "" if isinstance(val, (datetime, pd.Timestamp)): return val.strftime("%Y-%m-%dT%H:%M:%S") try: dt = datetime.strptime(str(val).strip()[:19], "%d.%m.%Y %H:%M:%S") return dt.strftime("%Y-%m-%dT%H:%M:%S") except: return str(val) def load_term_mapping_list(): if not os.path.exists(MAPPING_FILE): return [] try: with open(MAPPING_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: st.error(f"Ошибка при загрузке локального файла маппинга: {e}") return [] def load_term_mapping_dict(): mapping = {} data = load_term_mapping_list() for item in data: orig = str(item.get("original", "")) repl = str(item.get("replaced", "")) if orig and repl: mapping[orig] = repl return mapping def apply_term_mapping(term_number, mapping): if not mapping: return term_number return mapping.get(term_number, term_number) def create_xml_for_pp_memory(df_group, pp_number, pp_date, bik_code, term_mapping): org_name = str(df_group['Наименование юридического лица'].iloc[0]) org_inn = clean_id(df_group['ИНН'].iloc[0]) org_account = str(df_group['Номер расчётного счёта'].iloc[0]) root = ET.Element("ДанныеПоЗапросу") root.set("ДатаФормирования", datetime.now().strftime("%Y-%m-%dT%H:%M:%S")) root.set("НаименованиеОрганизации", org_name) root.set("ИНН", org_inn) root.set("БИК", bik_code) root.set("РасчетныйСчетОрганизации", org_account) root.set("ДатаПлатежногоПоручения", parse_date(pp_date)) root.set("НомерПлатежногоПоручения", clean_id(pp_number)) op_map = {'Покупка': 'оплата', 'Возврат': 'возврат', 'Отмена': 'отмена'} df_group['ДатаОперацииБезВремени'] = pd.to_datetime(df_group['Дата операции'], errors='coerce').dt.date grouped = df_group.groupby(['Номер мерчанта', 'ДатаОперацииБезВремени']) for (merchant, date_obj), group in grouped: if pd.isna(merchant) or pd.isna(date_obj): continue merchant_el = ET.SubElement(root, "СтрокаПоМерчанту") merchant_el.set("НомерМерчанта", clean_id(merchant)) merchant_el.set("Дата", date_obj.strftime("%Y-%m-%d")) control = {'оплата': {'cnt': 0, 'sum': 0.0}, 'возврат': {'cnt': 0, 'sum': 0.0}, 'отмена': {'cnt': 0, 'sum': 0.0}} for _, row in group.iterrows(): term_el = ET.SubElement(merchant_el, "СтрокаПоТерминалу") original_term = clean_id(row['Номер терминала']) mapped_term = apply_term_mapping(original_term, term_mapping) term_el.set("НомерТерминала", mapped_term) csv_op = str(row['Тип операции']).strip() xml_op = op_map.get(csv_op, 'оплата') amt = parse_decimal(row['Сумма операции']) comm = parse_decimal(row['Сумма комиссии']) settle = parse_decimal(row['Сумма расчета']) if xml_op in control: control[xml_op]['cnt'] += 1 control[xml_op]['sum'] += abs(amt) ET.SubElement(term_el, "УникальныйНомерТранзакции").text = clean_id(row['RRN']) ET.SubElement(term_el, "ВремяОперации").text = parse_datetime(row['Локальная дата операции']) ET.SubElement(term_el, "СуммаОперации").text = f"{amt:.2f}" ET.SubElement(term_el, "Комиссия").text = f"{comm:.2f}" ET.SubElement(term_el, "Возмещение").text = f"{settle:.2f}" ET.SubElement(term_el, "НомерКарты").text = str(row['Номер карты']) ET.SubElement(term_el, "СистемаРасчетов").text = str(row['Платежная система']) ET.SubElement(term_el, "КодАвторизации").text = clean_id(row['Код авторизации']) ET.SubElement(term_el, "ТипОперации").text = xml_op curr = str(row['Валюта']).strip() if curr == "RUR": curr = "RUB" ET.SubElement(term_el, "Валюта").text = curr ctrl_el = ET.SubElement(merchant_el, "КонтрольнаяСуммаПоДате") ET.SubElement(ctrl_el, "Дата").text = date_obj.strftime("%Y-%m-%d") for op, data in control.items(): if data['cnt'] > 0: op_el = ET.SubElement(ctrl_el, "Операция") op_el.set("ТипОперации", op) ET.SubElement(op_el, "КоличествоЗаписей").text = str(data['cnt']) ET.SubElement(op_el, "СуммаИтого").text = f"{data['sum']:.2f}" xml_str = ET.tostring(root, encoding='utf-8') dom = xml.dom.minidom.parseString(xml_str) pretty = dom.toprettyxml(indent=" ") safe_pp_num = re.sub(r'[\\/*?:"<>|]', '_', clean_id(pp_number)) pp_date_str = parse_date(pp_date) file_name = f"ПП_{safe_pp_num}_{pp_date_str}.xml" zip_path = f"XML_по_ПП/{pp_date_str}/{file_name}" return zip_path, pretty def run(plugin_info): with st.container(border=True): col1, col2 = st.columns([2, 1]) with col1: excel_file = st.file_uploader("Реестр (Excel .xlsx)", type=["xlsx"]) with col2: bik_code = st.text_input("БИК", value="045354601") with st.expander("Детальные настройки (Маппинг терминалов)"): st.write("Вы можете добавлять, изменять и удалять правила замены номеров терминалов.") mapping_list = load_term_mapping_list() df_mapping = pd.DataFrame(mapping_list) if df_mapping.empty: df_mapping = pd.DataFrame(columns=["original", "replaced"]) edited_df = st.data_editor( df_mapping, num_rows="dynamic", use_container_width=True, column_config={ "original": st.column_config.TextColumn("Оригинальный номер", required=True), "replaced": st.column_config.TextColumn("На что заменить", required=True) } ) if st.button("Сохранить изменения маппинга"): valid_df = edited_df.dropna(how="all") valid_df = valid_df[valid_df["original"].str.strip() != ""] new_mapping = valid_df.to_dict(orient="records") try: with open(MAPPING_FILE, "w", encoding="utf-8") as f: json.dump(new_mapping, f, ensure_ascii=False, indent=2) st.success("Настройки маппинга успешно сохранены в локальный файл!") except Exception as e: st.error(f"Ошибка при сохранении: {e}") if excel_file is not None: if st.button("Конвертировать", type="primary"): with st.spinner("Обработка данных..."): try: term_mapping = load_term_mapping_dict() df = pd.read_excel(excel_file, dtype={ 'Номер мерчанта': str, 'Номер терминала': str, 'RRN': str, 'Код авторизации': str, 'Номер карты': str, 'Номер ПП': str }) df.columns = [str(c).strip() for c in df.columns] required = ['Номер ПП', 'Дата ПП', 'Наименование юридического лица', 'ИНН', 'Номер расчётного счёта'] for col in required: if col not in df.columns: st.error(f"В файле отсутствует колонка: {col}") return df = df.dropna(subset=['Номер ПП', 'Дата ПП']) if df.empty: st.warning("Нет строк с заполненными номером и датой ПП") return grouped = df.groupby(['Номер ПП', 'Дата ПП']) zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: for (pp_num, pp_date), group in grouped: zip_path, xml_content = create_xml_for_pp_memory(group, pp_num, pp_date, bik_code, term_mapping) zf.writestr(zip_path, xml_content.encode('utf-8')) st.success(f"Успешно обработано платежных поручений: {len(grouped)}") st.download_button( label="Скачать ZIP-архив с XML", data=zip_buffer.getvalue(), file_name="XML_результаты.zip", mime="application/zip" ) except Exception as e: st.error(f"Ошибка при обработке файла: {e}")