#!/usr/bin/env python3 """ Получение документов из Bitrix24 iblock_57. Использование: python3 fetch_api.py # текущая неделя → data/api_records.json python3 fetch_api.py --all # все записи → data/raw_all.json python3 fetch_api.py --all --from 01.01.2026 # с даты → data/raw_all.json Использует B24 batch API: 50 запросов × 50 элементов = 2500 за 1 HTTP-вызов. Сырые данные сохраняются до парсинга — при сбое повторный запрос не нужен. """ import json import os import re import requests import argparse from datetime import datetime, timedelta WEBHOOK = 'https://engeocentr.bitrix24.ru/rest/4391/n8rq6nyzr9z8tvdz/' IBLOCK_TYPE = 'bitrix_processes' IBLOCK_ID = '57' BATCH_SIZE = 50 # B24 batch: максимум 50 команд за запрос PAGE_SIZE = 50 # lists.element.get: максимум 50 элементов за страницу SELECT = [ 'ID', 'NAME', 'DATE_CREATE', 'CREATED_BY', 'PROPERTY_1367', 'PROPERTY_423', 'PROPERTY_389', 'PROPERTY_391', 'PROPERTY_383', 'PROPERTY_421', 'PROPERTY_401', 'PROPERTY_427', 'PROPERTY_387', 'PROPERTY_397', 'PROPERTY_413', 'PROPERTY_415', 'PROPERTY_451', 'PROPERTY_503', 'PROPERTY_2185', 'PROPERTY_1415', 'PROPERTY_431', 'PROPERTY_2989', ] # ── Маппинги ────────────────────────────────────────────────────────────────── PAYER_MAP = {'373': 'БДК', '375': 'ТЗК', '377': 'ИНГЕО', '1059': 'ВЦПО'} OBJECT_MAP = { '1071': 'Камергер (общий)', '1073': 'Дмитровка (общий)', '15881': 'БД-ПА', '15883': 'КА-НС', '15885': 'КА-ПА', '17363': 'Стр.7 Б. Дмитровка, дом.9', '39129': 'БД-НС', } STATUS_MAP = { '363': 'Отклонено', '365': 'Согласовано', '367': 'Согласовано и распечатано', '419': 'Не установлен', } TYPE_MAP = {'357': 'Договор/ДС', '361': 'Счет/Акт'} DOC_PATTERNS = [ ('КС-2/КС-3', r'КС-2\s*/\s*КС-3'), ('УПД', r'\bУПД\b'), ('Счет-фактура', r'[Сс]чет-фактура|[Сс]ч-ф\b'), ('Акт', r'\bАкт\b'), ('Счет', r'[Сс]чет\s*№'), ('КС-2', r'КС-2\b'), ('КС-3', r'КС-3\b'), ('Справка', r'[Сс]правка'), ('Договор', r'Дог(?:овор)?\.?\s*№'), ('ДС', r'\bДС\s*№'), ] # ── Утилиты ─────────────────────────────────────────────────────────────────── def scalar(field): if not field: return '' if isinstance(field, dict): vals = list(field.values()) return vals[0] if vals else '' return field def multi_values(field): if not field: return [] if isinstance(field, dict): vals = list(field.values()) if vals and isinstance(vals[0], list): return vals[0] return [str(v) for v in vals if v] if isinstance(field, list): return field return [str(field)] def parse_doc_types(text): found = [] for name, pat in DOC_PATTERNS: if re.search(pat, text or ''): found.append(name) return list(dict.fromkeys(found)) def extract_contract_refs(text): refs = [] for m in re.findall(r'к\s*Договору\s*№\s*([\w\d\.\-\/]+)', text or ''): refs.append('Договор №' + m) for m in re.findall(r'по\s*ДС\s*№\s*([\w\d\.]+)', text or ''): refs.append('ДС №' + m) return refs def get_analysis_text(item): res = scalar(item.get('PROPERTY_2989')) if isinstance(res, dict): return res.get('TEXT', '') return res or '' def build_record(item: dict) -> dict: nom_dat = scalar(item.get('PROPERTY_391', '')) title = item.get('NAME', '') file_ids = [] for prop in ['PROPERTY_415', 'PROPERTY_451', 'PROPERTY_503', 'PROPERTY_2185']: file_ids += multi_values(item.get(prop)) file_ids = [f for f in file_ids if f] payer_id = str(scalar(item.get('PROPERTY_427', ''))) status_id = str(scalar(item.get('PROPERTY_401', ''))) type_id = str(scalar(item.get('PROPERTY_389', ''))) doc_types = parse_doc_types(nom_dat) or parse_doc_types(title) return { 'element_id': item['ID'], 'name': title, 'date_create': item.get('DATE_CREATE', '')[:10], 'contractor': scalar(item.get('PROPERTY_1367', '')), 'company_id': scalar(item.get('PROPERTY_423', '')), 'payer': PAYER_MAP.get(payer_id, payer_id), 'doc_type_raw': TYPE_MAP.get(type_id, type_id), 'doc_types': doc_types, 'contract_refs': extract_contract_refs(nom_dat), 'num_date_str': nom_dat, 'num_date_in': scalar(item.get('PROPERTY_383', '')), 'amount': scalar(item.get('PROPERTY_421', '')), 'status': STATUS_MAP.get(status_id, status_id), 'object': OBJECT_MAP.get(str(scalar(item.get('PROPERTY_387', ''))), scalar(item.get('PROPERTY_387', ''))), 'expense_item': scalar(item.get('PROPERTY_397', '')), 'comment': scalar(item.get('PROPERTY_413', '')), 'payment_due': scalar(item.get('PROPERTY_1415', '')), 'approved_date': scalar(item.get('PROPERTY_431', '')), 'file_ids': file_ids, 'file_count': len(file_ids), 'has_analysis': bool(get_analysis_text(item)), 'analysis_text': get_analysis_text(item), } # ── B24 API ─────────────────────────────────────────────────────────────────── def get_total(filter_val: dict) -> int: """Получить общее количество элементов.""" resp = requests.post(WEBHOOK + 'lists.element.get', json={ 'IBLOCK_TYPE_ID': IBLOCK_TYPE, 'IBLOCK_ID': IBLOCK_ID, 'FILTER': filter_val, 'LIST_NUM': 0, 'SELECT': ['ID'], }, timeout=30) resp.raise_for_status() return int(resp.json().get('total', 0)) def fetch_all_batch() -> list: """ Загрузка ВСЕХ активных элементов через B24 batch. 50 команд × 50 элементов = 2500 записей за 1 HTTP-запрос. Используем start= (не LIST_NUM) и cmd как список — как в референсе. """ total = get_total({'ACTIVE': 'Y'}) print(f'Всего элементов: {total}') select_str = '&'.join(f'select[{i}]={s}' for i, s in enumerate(SELECT)) filter_str = 'filter[ACTIVE]=Y' iblock_str = f'IBLOCK_TYPE_ID={IBLOCK_TYPE}&IBLOCK_ID={IBLOCK_ID}' order_str = 'ELEMENT_ORDER[DATE_CREATE]=DESC' all_items = [] batch_params = {'halt': 0, 'cmd': []} batch_num = 0 for start in range(0, total + PAGE_SIZE, PAGE_SIZE): cmd = ( f'lists.element.get?{iblock_str}' f'&{order_str}' f'&{filter_str}' f'&{select_str}' f'&start={start}' ) batch_params['cmd'].append(cmd) if len(batch_params['cmd']) >= 50 or start >= total: batch_num += 1 resp = requests.post(WEBHOOK + 'batch', json=batch_params, timeout=120) resp.raise_for_status() result = resp.json().get('result', {}).get('result', []) for page_items in result: if isinstance(page_items, list): all_items += page_items print(f' Батч {batch_num}: +{len(all_items) - (batch_num - 1) * 50 * 50} ' f'(итого {len(all_items)}/{total})') batch_params['cmd'] = [] return all_items def fetch_period(filter_val: dict) -> list: """ Загрузка за период через прямую пагинацию. Для недели/фильтров с датами — фильтры корректно передаются в POST-теле. """ all_items, start, page = [], 0, 1 while True: resp = requests.post(WEBHOOK + 'lists.element.get', json={ 'IBLOCK_TYPE_ID': IBLOCK_TYPE, 'IBLOCK_ID': IBLOCK_ID, 'FILTER': filter_val, 'ELEMENT_ORDER': {'DATE_CREATE': 'DESC'}, 'SELECT': SELECT, 'start': start, }, timeout=30) resp.raise_for_status() data = resp.json() items = data.get('result', []) if not items: break all_items += items total = int(data.get('total', 0)) print(f' Страница {page}: +{len(items)} (итого {len(all_items)}/{total})') if len(all_items) >= total: break start += PAGE_SIZE page += 1 return all_items # ── Main ────────────────────────────────────────────────────────────────────── def main(): ap = argparse.ArgumentParser() ap.add_argument('--all', action='store_true', help='Все записи (не только неделя)') ap.add_argument('--from', dest='date_from', help='Дата начала dd.mm.yyyy') args = ap.parse_args() os.makedirs('data', exist_ok=True) if args.all or args.date_from: # ── Полная выгрузка ─────────────────────────────────────────────── filter_val = {'ACTIVE': 'Y'} if args.date_from: filter_val['>=DATE_CREATE'] = args.date_from + ' 00:00:00' raw_path = 'data/raw_all.json' # Если файл уже есть — спрашиваем if os.path.exists(raw_path): with open(raw_path) as f: existing = json.load(f) print(f'⚠️ {raw_path} уже существует ({len(existing)} сырых записей)') ans = input('Перезаписать? [y/N]: ').strip().lower() if ans != 'y': print('Используем существующий файл.') records = [build_record(i) for i in existing] else: print('Загружаю из API...') raw_items = fetch_all(filter_val) with open(raw_path, 'w', encoding='utf-8') as f: json.dump(raw_items, f, ensure_ascii=False, indent=2) print(f'✅ Сырые данные → {raw_path} ({len(raw_items)} записей)') records = [build_record(i) for i in raw_items] else: print('Загружаю из API (batch)...') raw_items = fetch_all_batch() with open(raw_path, 'w', encoding='utf-8') as f: json.dump(raw_items, f, ensure_ascii=False, indent=2) print(f'✅ Сырые данные → {raw_path} ({len(raw_items)} записей)') records = [build_record(i) for i in raw_items] out_path = 'data/all_records.json' else: # ── Текущая неделя ──────────────────────────────────────────────── today = datetime.now() monday = today - timedelta(days=today.weekday()) date_from = monday.strftime('%d.%m.%Y') + ' 00:00:00' date_to = today.strftime('%d.%m.%Y') + ' 23:59:59' print(f'Период: {date_from} → {date_to}') filter_val = { '>=DATE_CREATE': date_from, '<=DATE_CREATE': date_to, } raw_items = fetch_period(filter_val) records = [build_record(i) for i in raw_items] out_path = 'data/api_records.json' with open(out_path, 'w', encoding='utf-8') as f: json.dump(records, f, ensure_ascii=False, indent=2) # Статистика print(f'\n✅ Записи → {out_path} ({len(records)})') print(f' С файлами: {sum(1 for r in records if r["file_count"] > 0)}') print(f' С анализом: {sum(1 for r in records if r["has_analysis"])}') print(f' С реф. договора: {sum(1 for r in records if r["contract_refs"])}') print(f' Без типа: {sum(1 for r in records if not r["doc_types"])}') from collections import Counter ct = Counter(t for r in records for t in r['doc_types']) print('\n Типы документов:') for t, n in ct.most_common(): print(f' {n:4}x {t}') if __name__ == '__main__': main()