#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ contact_lib.py — 全源联系人聚合入库模块 (2026-09-10 设计定稿) ============================================================ 统一联系人库 contact.db: 从 CCPC(ccpc.db) + ZC(ZC.db) 两源聚合联系人, 按 company+contact_name 去重合并(要素并集, 项目累加), 全要素保留。 设计(用户 2026-09-10 拍板): - 保留联系人归属公司; 可查"某公司所有联系人" - 全要素: 职责(role/position/department) + 座机(landline) + 手机(phone) + 邮箱(email) - 同公司内去重, 不同公司不去重; 跨源同公司同人合并(要素并集) - 由 ccpc_zc_sync.py Phase3 调用(编排器), 也可 CLI 单跑 Schema: contacts(id, company, contact_name, department, position, role, phone, landline, email, address, remarks) contact_projects(contact_id, source, project_key, project_name) 行为: - 全量重建(源库权威), 幂等 - 结果与现库完全一致时静默(exit 0 无输出) —— 配合编排器"无变化静默"语义 用法: python3 contact_lib.py # 重建并写库 python3 contact_lib.py force # 同(默认即全量) """ import os, re, sys, sqlite3, hashlib, json CONTACT_DB = "/root/contact.db" CCPC_DB = "/root/ccpc.db" ZC_DB = "/root/ZC.db" PHONE_NORM = re.compile(r'[\s\-—–_()（）]+') MOBILE_RE = re.compile(r'^1[3-9]\d{9}$') def norm_phone(v): if not v: return '' return PHONE_NORM.sub('', str(v)).strip() def clean_landline(v, phone_val=''): """座机清洗(2026-09-10): 源数据 99% 的 'landline' 是手机号重复存 → 手机格式(1[3-9]+9位) 或 与手机同号 的值一律剔除; 多值逐项过滤; 真座机(区号开头)保留""" if not v: return '' keep = [] for part in str(v).split('/'): p = norm_phone(part) if not p: continue if MOBILE_RE.match(p): continue # 手机格式 → 不是座机 if norm_phone(phone_val) and p == norm_phone(phone_val): continue # 与手机同号 → 重复 if p not in keep: keep.append(p) return ' / '.join(keep) def merge_val(a, b): """字段并集: 非空值去重, 多值用 ' / ' 连接""" vals = [x.strip() for x in (a or '').split(' / ') if x.strip()] if b and b.strip() not in vals: vals.append(b.strip()) return ' / '.join(vals) class ContactMerger: def __init__(self): self.by_key = {} # (company, contact_name) -> {fields..., projects:[]} def add(self, company, name, project, **fields): company = (company or '').strip() name = (name or '').strip() if not company or not name: return False key = (company, name) rec = self.by_key.get(key) if rec is None: rec = {'company': company, 'contact_name': name, 'department': '', 'position': '', 'role': '', 'phone': '', 'landline': '', 'email': '', 'address': '', 'remarks': '', 'projects': []} self.by_key[key] = rec for f in ('department', 'position', 'role', 'address', 'remarks'): if fields.get(f): rec[f] = merge_val(rec[f], fields[f]) # 电话归一: 手机 phone 与 座机 landline 分开并集(landline 先过真座机清洗) if fields.get('phone'): rec['phone'] = merge_val(rec['phone'], norm_phone(fields['phone'])) if fields.get('landline'): _ln = clean_landline(fields['landline'], fields.get('phone') or '') if _ln: rec['landline'] = merge_val(rec['landline'], _ln) if fields.get('email'): rec['email'] = merge_val(rec['email'], (fields['email'] or '').strip()) if project and project not in rec['projects']: rec['projects'].append(project) return True def load_ccpc(m): """CCPC 源: cceup_contacts + cceup_projects(id→project_id/project_name)""" con = sqlite3.connect(CCPC_DB) con.row_factory = sqlite3.Row rows = con.execute(""" SELECT ct.project_id AS fk, ct.company, ct.contact_name, ct.phone, ct.landline, ct.email, ct.role, ct.address FROM cceup_contacts ct """).fetchall() pmap = {r['id']: (r['project_id'], r['project_name']) for r in con.execute( "SELECT id, project_id, project_name FROM cceup_projects")} con.close() n = 0 for r in rows: pid, pname = pmap.get(r['fk'], ('', '')) if m.add(r['company'], r['contact_name'], ('ccpc', pid, pname), department='', position='', role=r['role'], phone=r['phone'], landline=r['landline'], email=r['email'], address=r['address'], remarks=''): n += 1 return n def load_zc(m): con = sqlite3.connect(ZC_DB) con.row_factory = sqlite3.Row rows = con.execute(""" SELECT ct.project_id AS fk, ct.company, ct.contact_name, ct.department, ct.position, ct.phone, ct.role, ct.address, ct.remarks FROM zc_contacts ct """).fetchall() pmap = {r['project_id']: r['project_name'] for r in con.execute( "SELECT project_id, project_name FROM zc_projects")} con.close() n = 0 for r in rows: pname = pmap.get(r['fk'], '') if m.add(r['company'], r['contact_name'], ('zc', r['fk'], pname), department=r['department'], position=r['position'], role=r['role'], phone=r['phone'], landline='', email='', address=r['address'], remarks=r['remarks']): n += 1 return n def snapshot(merger=None): """统一快照: [[10字段..., [[source,key,name],...]], ...] 排序。merger=None 时读现库""" if merger is not None: out = [] for key in sorted(merger.by_key): r = merger.by_key[key] out.append([r['company'], r['contact_name'], r['department'], r['position'], r['role'], r['phone'], r['landline'], r['email'], r['address'], r['remarks'], sorted([list(p) for p in r['projects']])]) return out if not os.path.exists(CONTACT_DB): return None try: con = sqlite3.connect(CONTACT_DB) rows = con.execute("""SELECT company, contact_name, department, position, role, phone, landline, email, address, remarks FROM contacts ORDER BY company, contact_name""").fetchall() cps = con.execute("""SELECT contact_id, source, project_key, project_name FROM contact_projects ORDER BY contact_id""").fetchall() proj = {} for cid, s, pk, pn in cps: proj.setdefault(cid, []).append([s, pk, pn]) ids = con.execute("SELECT id FROM contacts ORDER BY company, contact_name").fetchall() out = [] for i, row in enumerate(rows): cid = ids[i][0] out.append([list(row) + [sorted(proj.get(cid, []))]]) con.close() return out except Exception: return None def content_hash(items): if items is None: return None return hashlib.sha256(json.dumps(items, ensure_ascii=False).encode()).hexdigest() def write_db(merger): new_items = snapshot(merger) new_h = content_hash(new_items) cur_h = content_hash(snapshot(None)) if cur_h == new_h: return False # 无变化, 不落盘 # 全量重建 con = sqlite3.connect(CONTACT_DB) c = con.cursor() c.execute("DROP TABLE IF EXISTS contact_projects") c.execute("DROP TABLE IF EXISTS contacts") c.execute("""CREATE TABLE contacts ( id INTEGER PRIMARY KEY AUTOINCREMENT, company TEXT NOT NULL, contact_name TEXT NOT NULL DEFAULT '', department TEXT DEFAULT '', position TEXT DEFAULT '', role TEXT DEFAULT '', phone TEXT DEFAULT '', landline TEXT DEFAULT '', email TEXT DEFAULT '', address TEXT DEFAULT '', remarks TEXT DEFAULT '')""") c.execute("CREATE UNIQUE INDEX idx_contacts_company_name ON contacts(company, contact_name)") c.execute("""CREATE TABLE contact_projects ( contact_id INTEGER NOT NULL, source TEXT NOT NULL, project_key TEXT NOT NULL, project_name TEXT DEFAULT '')""") c.execute("CREATE INDEX idx_cp_contact ON contact_projects(contact_id)") for key in sorted(merger.by_key): r = merger.by_key[key] c.execute( "INSERT INTO contacts (company, contact_name, department, position, role, phone," " landline, email, address, remarks) VALUES (?,?,?,?,?,?,?,?,?,?)", (r['company'], r['contact_name'], r['department'], r['position'], r['role'], r['phone'], r['landline'], r['email'], r['address'], r['remarks'])) cid = c.lastrowid c.executemany( "INSERT INTO contact_projects (contact_id, source, project_key, project_name) VALUES (?,?,?,?)", [(cid, s, pk, pn) for s, pk, pn in r['projects']]) con.commit() n = c.execute("SELECT COUNT(*) FROM contacts").fetchone()[0] nc = c.execute("SELECT COUNT(*) FROM contact_projects").fetchone()[0] con.close() return (n, nc) def main(): m = ContactMerger() n_ccpc = load_ccpc(m) n_zc = load_zc(m) src_rows = n_ccpc + n_zc result = write_db(m) if result is False: return 0 # 静默 n_contacts, n_proj = result n_company = len({k[0] for k in m.by_key}) print(f"联系人库: 源行 {src_rows} (ccpc {n_ccpc} / zc {n_zc}) → 去重后 {n_contacts} 人 / " f"{n_company} 公司 / {n_proj} 项目关联") return 0 if __name__ == "__main__": sys.exit(main())