#!/usr/bin/env python3 """ Parse 中项网 (CCPC) .docx and .xls/.xlsx files into ccpc.db. For DOCX: convert document to clean text, extract contacts into cceup_contacts. For XLS: extract structured fields + contacts into cceup_contacts. Usage: python3 parse_ccpc.py # Incremental: new files python3 parse_ccpc.py force # Force re-process all python3 parse_ccpc.py test # Test: 3 files """ import sys, json, os, re, shutil, urllib.parse import requests from requests.auth import HTTPBasicAuth from xml.etree import ElementTree as ET import sqlite3 from datetime import datetime from openpyxl import load_workbook from docx import Document # ─── CONFIG ──────────────────────────────────────────────── WEBDAV_USER = 'chniir0000@outlook.com' WEBDAV_PASS = 'REDACTED_SEE_DOTENV' WEBDAV_BASE = 'https://dav.jianguoyun.com' DOWNLOAD_DIRS = ['/dav/Downloads/', '/dav/%E4%B8%8B%E8%BD%BD1/'] LOCAL_CACHE = '/data/ccpc_source' DB_PATH = '/root/ccpc.db' TRACKING_FILE = '/data/ccpc_processed.txt' os.makedirs(LOCAL_CACHE, exist_ok=True) auth = HTTPBasicAuth(WEBDAV_USER, WEBDAV_PASS) # ─── XLS FIELD MAPPING ───────────────────────────────────── XLS_FIELD_MAP = { '项目名称': 'project_name', '项目类型': 'project_type', '项目ID': 'project_id', '项目所属行业': 'industry', '所属领域类型': 'field_type', '所属省份': 'province', '所属地级市': 'city', '进展阶段': 'phase', '发布时间': 'publish_date', '跟踪版本号': 'version', '项目性质': 'nature', '预算投资总额(万元)': 'budget', '投资性质': 'invest_nature', '资金到位情况': 'funding', '建设等级': 'grade', '预计开建时间': 'start_time', '预计截止时间': 'end_time', '设备来源': 'equipment_source', '建筑面积': 'building_area', '占地面积': 'land_area', '有无钢结构': 'steel_structure', '所需材料设备': 'equipment_list', '设备采购情况': 'equipment_list', '项目进展': 'progress', '项目概况': 'overview', '项目详情': 'detail', '业主单位': 'owner_company', # DOCX tracking version fields '项目标题': 'project_name', '更新版本': 'version', '更新时间': 'publish_date', '所属栏目': 'project_type', '建筑层数': 'building_floors', '供暖方式': 'heating_method', '外墙材料': 'wall_material', '装修': 'decoration', '有无空调': 'has_ac', '有无立体停车位': 'has_parking', '有无电梯': 'has_elevator', '项目所在地': 'project_address', '专题标签': 'tags', '工艺流程': 'process_flow', '主体施工阶段': 'phase_construction', '主体设计阶段': 'phase_design', '商机描述': 'opportunity', '关键工程': 'key_works', '具体需求情况': 'procurement_needs', # Contact source fields (parsed into cceup_contacts) '业主单位联系人以及联系方式': '__contacts_owner__', '设计院联系人以及联系方式': '__contacts_design__', '施工单位联系人以及联系方式': '__contacts_contractor__', '参考联系人': '__contacts_ref__', } ALL_PROJECT_FIELDS = [ 'project_name', 'project_type', 'project_id', 'industry', 'field_type', 'province', 'city', 'phase', 'publish_date', 'version', 'nature', 'budget', 'invest_nature', 'funding', 'grade', 'start_time', 'end_time', 'equipment_source', 'building_area', 'land_area', 'steel_structure', 'equipment_list', 'progress', 'opportunity', 'key_works', 'overview', 'detail', 'owner_company', 'contacts', 'raw_json', 'building_floors', 'heating_method', 'wall_material', 'decoration', 'has_ac', 'has_parking', 'has_elevator', 'project_address', 'tags', 'process_flow', 'phase_construction', 'phase_design', 'procurement_equipment', 'procurement_needs' ] # Contact regex patterns PHONE_RE = re.compile(r'1[3-9]\d{9}') LANDLINE_RE = re.compile(r'0\d{2,3}-?\d{7,8}') NAME_RE = re.compile(r'联系人[：:]\s*([\u4e00-\u9fff·]{2,6})') COMPANY_RE = re.compile(r'^(.+?)[(（]') def log(msg): print(f'[{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}] {msg}') def webdav_list(remote_dir): url = WEBDAV_BASE + remote_dir r = requests.request('PROPFIND', url, auth=auth, headers={'Depth': '1'}, timeout=30) if r.status_code != 207: return [] root = ET.fromstring(r.content) ns = {'d': 'DAV:'} files = [] for resp in root.findall('.//d:response', ns): href_el = resp.find('d:href', ns) if href_el is None: continue fn = (href_el.text or '').rstrip('/').rsplit('/', 1)[-1] # 解码 URL 编码的文件名（中文等） fn = urllib.parse.unquote(fn) if fn: files.append(fn) return files def webdav_download(remote_base, filename, local_path): url = WEBDAV_BASE + remote_base + filename r = requests.get(url, auth=auth, timeout=60) if r.status_code == 200: with open(local_path, 'wb') as f: f.write(r.content) return True return False def title_from_filename(filename): name = re.sub(r'-\d+\.(docx|xls|xlsx)$', '', filename) name = re.sub(r'\.(docx|xls|xlsx)$', '', name) name = re.sub(r'\s*\(\d+\)\s*$', '', name) try: name = urllib.parse.unquote(name) except Exception: pass return name # ─── CONTACT PARSER ──────────────────────────────────────── def parse_contacts_from_text(text_blob, role_label=''): """Parse contact text block into list of contact dicts.""" contacts = [] if not text_blob or not text_blob.strip(): return contacts # Split by company blocks (look for pattern: company name followed by (业主)/(设计)/(施工) etc.) # or by double newlines, or by consecutive company-like lines lines = text_blob.strip().split('\n') # Group lines into contact blocks # A contact block starts with a company name (no "联系人"/"联系方式"/"地址" prefix) blocks = [] current_block = [] for line in lines: line = line.strip() if not line: continue # Check if this line starts a new contact block (company name, not a "联系人"/"电话" continuation) if (not current_block or re.match(r'^[\u4e00-\u9fff]', line) and not line.startswith(('联系人', '联系方式', '手机', '地址', '电话', '姓名', '部门')) and len(line) > 4 and ('有限公司' in line or '(业主)' in line or '(设计' in line or '(施工' in line or '(法人' in line or '(项目' in line or '(股东' in line or '公司' in line or '厂' in line or '院' in line)): if current_block: blocks.append('\n'.join(current_block)) current_block = [line] else: current_block.append(line) if current_block: blocks.append('\n'.join(current_block)) for block in blocks: block = block.strip() if not block or '仅供参考' in block or '版权所有' in block: continue lines_b = block.split('\n') # First line is company name + role first_line = lines_b[0].strip() company = re.sub(r'\s*[(（][^)）]*[)）]\s*$', '', first_line).strip() if not company: continue # Join remaining lines for searching rest = '\n'.join(lines_b[1:]) # Find names names = NAME_RE.findall(block) phones = list(set(PHONE_RE.findall(block))) landlines = list(set(LANDLINE_RE.findall(block))) # Extract address addr = '' addr_m = re.search(r'地址[：:]\s*(.+)', block) if addr_m: addr = addr_m.group(1).strip() if names: for name in names: contacts.append({ 'company': company, 'contact_name': name, 'phone': ' / '.join(phones) if phones else '', 'landline': ' / '.join(landlines) if landlines else '', 'address': addr, 'role': role_label, }) elif phones: contacts.append({ 'company': company, 'contact_name': '', 'phone': ' / '.join(phones), 'landline': ' / '.join(landlines) if landlines else '', 'address': addr, 'role': role_label, }) return contacts # ─── XLS PARSER ──────────────────────────────────────────── def parse_xls(filepath): tmp = '/tmp/ccpc_xls_temp.xlsx' shutil.copy2(filepath, tmp) try: wb = load_workbook(tmp, read_only=True) ws = wb[wb.sheetnames[0]] headers, data = None, None for i, row in enumerate(ws.iter_rows(values_only=True)): if i == 0: headers = [str(v) if v else '' for v in row] elif i == 1: data = [str(v) if v else '' for v in row] break wb.close() if not headers or not data: return None, [] raw = dict(zip(headers, data)) record = {} contacts_raw = {} for cn, en in XLS_FIELD_MAP.items(): val = raw.get(cn, '') or '' if en.startswith('__contacts_'): contacts_raw[en.replace('__contacts_', '').replace('__', '')] = val else: record[en] = str(val) if not record.get('project_name'): record['project_name'] = title_from_filename(os.path.basename(filepath)) # Build clean detail from raw fields detail_lines = [] for k, v in raw.items(): if v and v.strip() and not k.startswith('__') and k not in ('业主单位联系人以及联系方式', '设计院联系人以及联系方式', '施工单位联系人以及联系方式', '参考联系人'): if k in ('项目概况', '项目详情', '项目进展', '所需材料设备', '设备采购情况', '项目所在地'): detail_lines.append(f'\n【{k}】\n{v.strip()}') record['detail'] = '\n'.join(detail_lines) record['raw_json'] = json.dumps(raw, ensure_ascii=False) # Parse contacts all_contacts = [] role_map = {'contacts_owner': '业主单位', 'contacts_design': '设计院', 'contacts_contractor': '施工单位', 'contacts_ref': '参考联系人'} for key, text in contacts_raw.items(): if text and text.strip(): role = role_map.get(key, key) all_contacts.extend(parse_contacts_from_text(text, role)) return record, all_contacts finally: if os.path.exists(tmp): os.remove(tmp) # ─── DOCX PARSER ─────────────────────────────────────────── SECTION_NAMES = {'所需材料设备', '设备采购情况', '项目进展', '项目概况', '项目详情', '业主单位联系人以及联系方式', '设计院联系人以及联系方式', '施工单位联系人以及联系方式', '参考单位（仅供参考）', '目录', '项目跟踪版本', '项目基础信息'} def _apply_docx_field(record, cn_key, val): """Map DOCX Chinese key to English field and set on record.""" en = XLS_FIELD_MAP.get(cn_key) if en and not en.startswith('__contacts_'): if not record.get(en): record[en] = val def parse_docx(filepath): doc = Document(filepath) record = {} contacts_raw = {} # section_name -> text detail_parts = [] for table in doc.tables: rows_data = [] section_name = None for ri, row in enumerate(table.rows): cells = [c.text.strip() for c in row.cells] all_same = cells and all(c == cells[0] for c in cells) ct = cells[0] if cells else '' is_hdr = all_same and (ct in SECTION_NAMES or (ri > 0 and len(ct) < 25)) if is_hdr and ri > 0: # Process previous section if section_name and rows_data: if section_name == '项目基础信息': # Extract key-value pairs for clean text display + structured fields kv_lines = [] for rcells in rows_data: deduped = [] for c in rcells: if not deduped or c != deduped[-1]: deduped.append(c) for i in range(0, len(deduped), 2): if i + 1 < len(deduped): k, v = deduped[i].strip(), deduped[i+1].strip() if k and v: kv_lines.append(f'{k}：{v}') _apply_docx_field(record, k, v) if kv_lines: detail_parts.append('\n'.join(kv_lines)) elif section_name == '项目跟踪版本': if len(rows_data) >= 2: hdr, dat = rows_data[0], rows_data[1] for i in range(0, min(len(hdr), len(dat))): if hdr[i].strip() and dat[i].strip(): detail_parts.append(f'{hdr[i].strip()}：{dat[i].strip()}') _apply_docx_field(record, hdr[i].strip(), dat[i].strip()) elif section_name in ('项目详情', '项目概况'): if rows_data and rows_data[0]: val = rows_data[0][0] or '' if val and '版权所有' not in val: detail_parts.append(f'\n【{section_name}】\n{val}') if not record.get('overview') and section_name == '项目概况': record['overview'] = val if not record.get('detail') and section_name == '项目详情': record['detail'] = val elif section_name in ('所需材料设备', '设备采购情况', '项目进展'): if rows_data and rows_data[0]: val = rows_data[0][0] or '' if val and '版权所有' not in val: detail_parts.append(f'\n【{section_name}】\n{val}') if not record.get('equipment_list') and section_name in ('所需材料设备', '设备采购情况'): record['equipment_list'] = val if not record.get('progress') and section_name == '项目进展': record['progress'] = val elif '联系人' in section_name: if rows_data and rows_data[0]: val = rows_data[0][0] or '' if val and '仅供参考' not in val and '版权所有' not in val: contacts_raw[section_name] = val rows_data = [] section_name = ct elif not is_hdr: rows_data.append(cells) # Last section if section_name and rows_data: if section_name == '项目基础信息': kv_lines = [] for rcells in rows_data: deduped = [] for c in rcells: if not deduped or c != deduped[-1]: deduped.append(c) for i in range(0, len(deduped), 2): if i + 1 < len(deduped): k, v = deduped[i].strip(), deduped[i+1].strip() if k and v: kv_lines.append(f'{k}：{v}') _apply_docx_field(record, k, v) if kv_lines: detail_parts.append('\n'.join(kv_lines)) elif section_name in ('项目详情', '项目概况'): if rows_data and rows_data[0]: val = rows_data[0][0] or '' if val and '版权所有' not in val: detail_parts.append(f'\n【{section_name}】\n{val}') if not record.get('overview') and section_name == '项目概况': record['overview'] = val if not record.get('detail') and section_name == '项目详情': record['detail'] = val elif section_name in ('所需材料设备', '设备采购情况', '项目进展'): if rows_data and rows_data[0]: val = rows_data[0][0] or '' if val and '版权所有' not in val: detail_parts.append(f'\n【{section_name}】\n{val}') if not record.get('equipment_list') and section_name in ('所需材料设备', '设备采购情况'): record['equipment_list'] = val if not record.get('progress') and section_name == '项目进展': record['progress'] = val elif '联系人' in section_name: if rows_data and rows_data[0]: val = rows_data[0][0] or '' if val and '仅供参考' not in val and '版权所有' not in val: contacts_raw[section_name] = val # Set detail record['detail'] = '\n\n'.join(detail_parts) # Name from first paragraph or heading for para in doc.paragraphs: text = para.text.strip() if not text: continue style = (para.style.name or '').lower() if 'heading' in style or 'title' in style: record['project_name'] = text break else: # Try first non-empty paragraph (often "Normal" style but actually a title) for para in doc.paragraphs: text = para.text.strip() if text and len(text) > 5 and not text.startswith('目录'): record['project_name'] = text break if not record.get('project_name'): record['project_name'] = title_from_filename(os.path.basename(filepath)) record['raw_json'] = '' # Parse contacts from raw contact sections all_contacts = [] role_map = { '业主单位联系人以及联系方式': '业主单位', '设计院联系人以及联系方式': '设计院', '施工单位联系人以及联系方式': '施工单位' } for section_name, text in contacts_raw.items(): role = role_map.get(section_name, '') all_contacts.extend(parse_contacts_from_text(text, role)) return record, all_contacts # ─── DB ──────────────────────────────────────────────────── def upsert_project(record): db = sqlite3.connect(DB_PATH) pid, pname = record.get('project_id', ''), record.get('project_name', '') existing = None if pid: existing = db.execute('SELECT id FROM cceup_projects WHERE project_id = ?', (pid,)).fetchone() if existing is None and pname: existing = db.execute('SELECT id FROM cceup_projects WHERE project_name = ?', (pname,)).fetchone() vals = [record.get(f, '') for f in ALL_PROJECT_FIELDS] if existing: set_c = ', '.join(f'{f}=?' for f in ALL_PROJECT_FIELDS) db.execute(f'UPDATE cceup_projects SET {set_c} WHERE id=?', vals + [existing[0]]) proj_id = existing[0] action = 'UPDATE' else: ph = ', '.join(['?'] * len(ALL_PROJECT_FIELDS)) db.execute(f'INSERT INTO cceup_projects ({",".join(ALL_PROJECT_FIELDS)}) VALUES ({ph})', vals) proj_id = db.execute('SELECT last_insert_rowid()').fetchone()[0] action = 'INSERT' db.commit() db.close() return action, proj_id def save_contacts(project_id, contacts): if not contacts: return 0 db = sqlite3.connect(DB_PATH) # Delete existing contacts for this project db.execute('DELETE FROM cceup_contacts WHERE project_id = ?', (project_id,)) saved = 0 for c in contacts: db.execute( 'INSERT INTO cceup_contacts (project_id, company, contact_name, phone, landline, address, role) VALUES (?,?,?,?,?,?,?)', (project_id, c.get('company', ''), c.get('contact_name', ''), c.get('phone', ''), c.get('landline', ''), c.get('address', ''), c.get('role', '')) ) saved += 1 db.commit() db.close() return saved # ─── MAIN ────────────────────────────────────────────────── def main(): mode = sys.argv[1] if len(sys.argv) > 1 else 'all' log(f'=== CCPC parse [mode={mode}] ===') # List files all_remote = [] for d in DOWNLOAD_DIRS: files = webdav_list(d) supported = [f for f in files if f.endswith(('.docx', '.xls', '.xlsx'))] log(f' {d}: {len(supported)} files') if supported: all_remote.extend((d, f) for f in supported) log(f'Total: {len(all_remote)} files') # Which to process processed = set() if os.path.exists(TRACKING_FILE): with open(TRACKING_FILE) as f: processed = set(l.strip() for l in f if l.strip()) if mode == 'test': seen = {} for d, f in all_remote: if f not in seen: seen[f] = d to_process = [(seen[f], f) for f in list(seen.keys())[:3]] elif mode == 'force': to_process = all_remote else: # Daily: 处理全部，upsert_project 按 project_name 去重 to_process = all_remote log(f'Daily: {len(all_remote)} total files (upsert dedup)') if not to_process: db = sqlite3.connect(DB_PATH) n = db.execute('SELECT COUNT(*) FROM cceup_projects').fetchone()[0] nc = db.execute('SELECT COUNT(*) FROM cceup_contacts').fetchone()[0] db.close() log(f'Nothing new. Projects: {n}, Contacts: {nc}') return # Process ins = upd = err = 0 for base_dir, fname in to_process: lp = os.path.join(LOCAL_CACHE, fname) if not os.path.exists(lp) and not webdav_download(base_dir, fname, lp): err += 1 continue try: if fname.endswith(('.xls', '.xlsx')): rec, contacts = parse_xls(lp) else: rec, contacts = parse_docx(lp) if rec and rec.get('project_name'): # 将联系人写入 record['contacts'] 字段 if contacts: contact_lines = [] for c in contacts: parts = [] if c.get('company'): parts.append(c['company']) if c.get('contact_name'): parts.append(f"联系人：{c['contact_name']}") if c.get('phone'): parts.append(f"联系方式：{c['phone']}") if c.get('landline'): parts.append(f"座机：{c['landline']}") if c.get('address'): parts.append(f"地址：{c['address']}") if c.get('role'): parts.append(f"角色：{c['role']}") contact_lines.append('\n'.join(parts)) rec['contacts'] = '\n\n'.join(contact_lines) act, proj_id = upsert_project(rec) c_saved = save_contacts(proj_id, contacts) if act == 'INSERT': ins += 1 else: upd += 1 if ins % 10 == 0 and ins > 0: log(f' Progress: +{ins} U{upd} err{err}') else: err += 1 except Exception as e: err += 1 log(f' ERROR {fname[:40]}: {e}') db = sqlite3.connect(DB_PATH) n = db.execute('SELECT COUNT(*) FROM cceup_projects').fetchone()[0] nc = db.execute('SELECT COUNT(*) FROM cceup_contacts').fetchone()[0] db.close() log(f'Done: +{ins} U{upd} err{err}') log(f'ccpc.db: {n} projects, {nc} contacts') if __name__ == '__main__': main()