Commit 7a75747e authored by wenwen.tang's avatar wenwen.tang 😕

feat(tools): 合同批量导入脚本;合同目录不入库

- tools/import_contracts.py:把「项目合约」里的合同挂到台账 AL 列
  默认干跑(只报匹配),--apply 才写入,--replace 覆盖(默认追加,一份可挂多份)
  解析文件名开头的单元号串:1101、1115 / 124.155号 / 132&227车位 / 501-502室 / 601A&702 / 169
  每个号再按台账实际存在的 n室 / n车位 定夺(都在时看「车位 / 号」线索)
  干跑结果:29 个文件 → 35 个单元,零未匹配、零歧义
- .gitignore:data/contracts/(运行时数据)、项目合约/(93MB + 商业条款,不入库)
parent d6fd7909
...@@ -21,6 +21,7 @@ data/send_log.json ...@@ -21,6 +21,7 @@ data/send_log.json
data/config.json.before-round3 data/config.json.before-round3
data/batches/ data/batches/
data/drafts/ data/drafts/
data/contracts/
data/attachments/ data/attachments/
logs/ logs/
...@@ -59,3 +60,6 @@ deploy/wecom/x/ ...@@ -59,3 +60,6 @@ deploy/wecom/x/
Thumbs.db Thumbs.db
desktop.ini desktop.ini
.DS_Store .DS_Store
# ---- 合同原件原始件(93MB,含商业条款,不入库;导入后落在 data/contracts/)----
项目合约/
# -*- coding: utf-8 -*-
"""把「项目合约」里的合同原件批量挂到台账「合同原件」列(AL)。
用法:
python tools/import_contracts.py # 干跑:只报匹配结果,不写任何东西
python tools/import_contracts.py --apply # 真正导入(上传文件 + 写台账)
python tools/import_contracts.py --dir 项目合约 --apply
python tools/import_contracts.py --replace # 已有合同的单元先清空再挂
匹配规则(从文件名开头解析,不看公司名里的数字):
开头的「单元号序列」才是单元:1101、1115 / 124.155 / 132&227 / 501-502 /
601A&702 / 169 …,遇到中文(室 / 号 / 车位 / 公司名)即停止。
每个单元号再按台账实际存在的写法定夺:`n室` 存在就挂房间,`n车位` 存在就挂车位;
两个都存在时看文件名线索(含「车位 / 号」→ 车位,含「室 / 房屋」→ 房间)。
写入口径:与页面一致 —— 同一个单元已有合同则**追加**(一份可挂多份:
租赁合同 + 补充协议 + 车位合同),`--replace` 则改为覆盖。
"""
import argparse
import base64
import io
import json
import os
import re
import sys
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
from rentos import config as C # noqa: E402
from rentos import contracts as CT # noqa: E402
from rentos import tables # noqa: E402
# 文件名开头:单元号(3~4 位数字 + 可选字母)用 、,,&+.~- 连接
_HEAD = re.compile(r"^\s*((?:\d{3,4}[A-Za-z]?)(?:\s*[、,,&+.~\-]\s*\d{3,4}[A-Za-z]?)*)")
_NUM = re.compile(r"\d{3,4}[A-Za-z]?")
def parse_units(filename):
"""文件名 → 候选单元号列表(只取开头那串)。"""
stem = os.path.splitext(os.path.basename(filename))[0]
m = _HEAD.match(stem)
return _NUM.findall(m.group(1)) if m else []
def park_hint(stem):
"""文件名给的房间/车位线索:True=车位 False=房间 None=没线索。"""
if "车位" in stem:
return True
if "室" in stem or "房屋" in stem:
return False
if "号" in stem:
return True # 「124.155号」这类
return None
def resolve(nums, stem, unit_map):
"""候选单元号 → [(原始号, 台账单元, 行 id 列表), ...];台账里没有的返回单元为 None。"""
hint = park_hint(stem)
out = []
for n in nums:
cands = [n + "室", n + "车位"]
hit = [u for u in cands if u in unit_map]
if not hit:
out.append((n, None, []))
continue
if len(hit) > 1 and hint is not None:
hit = [u for u in hit if (u.endswith("车位") == hint)] or hit
unit = hit[0]
out.append((n, unit, unit_map[unit]))
return out
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--dir", default="项目合约")
ap.add_argument("--apply", action="store_true", help="真正写入(默认干跑)")
ap.add_argument("--replace", action="store_true", help="覆盖已有合同(默认追加)")
args = ap.parse_args()
src = os.path.join(ROOT, args.dir)
if not os.path.isdir(src):
print("目录不存在:%s" % src)
return 2
t = tables.load_table(tables.RENT_KEY)
if not t:
print("台账尚未就绪")
return 2
unit_map = {}
for r in t["rows"]:
u = str(r.get("C") or "").strip()
if u:
unit_map.setdefault(u, []).append(r.get("_id"))
files = sorted(f for f in os.listdir(src) if not f.startswith("."))
plan, miss, ambiguous = [], [], []
for fn in files:
stem = os.path.splitext(fn)[0]
nums = parse_units(fn)
if not nums:
miss.append((fn, "文件名开头没有单元号"))
continue
for raw, unit, rids in resolve(nums, stem, unit_map):
if not unit:
miss.append((fn, "台账里没有 %s室 / %s车位" % (raw, raw)))
continue
if len(rids) > 1:
ambiguous.append((fn, unit, rids))
plan.append({"file": fn, "unit": unit, "row_id": rids[0]})
print("源目录:%s(%d 个文件)" % (src, len(files)))
print("=" * 96)
cur = None
for p in plan:
if p["file"] != cur:
cur = p["file"]
print("\n%s" % cur)
print(" → %s" % ", ".join(x["unit"] for x in plan if x["file"] == cur))
print("\n" + "=" * 96)
print("匹配上:%d 个文件 → %d 个单元" % (len({p["file"] for p in plan}), len(plan)))
if ambiguous:
print("\n⚠ 同一单元号命中多行(需人工确认,导入会取第一行):")
for fn, unit, rids in ambiguous:
print(" %s → %s : %s" % (fn, unit, rids))
if miss:
print("\n⚠ 未匹配:")
for fn, why in miss:
print(" %s —— %s" % (fn, why))
if not args.apply:
print("\n(干跑结束,未写入任何数据;确认无误后加 --apply 再跑一次)")
return 0
print("\n开始导入(%s)…" % ("覆盖" if args.replace else "追加"))
updates, done, failed = [], 0, []
for p in plan:
fp = os.path.join(src, p["file"])
try:
raw = open(fp, "rb").read()
meta = CT.save("data:application/octet-stream;base64," + base64.b64encode(raw).decode(),
p["file"], "合同导入")
except Exception as e: # noqa: BLE001
failed.append((p["file"], str(e)))
continue
row = [r for r in t["rows"] if r.get("_id") == p["row_id"]][0]
items = [] if args.replace else CT.parse(row.get("AL"))
items = [x for x in items if x["id"] != meta["id"]]
items.append({"id": meta["id"], "name": meta["name"]})
updates.append((p["row_id"], "AL", CT.dump(items)))
done += 1
if updates:
res = tables.update_cells(tables.RENT_KEY, updates, who="合同导入")
print("已写入 %d 处(其中内容有变化 %d 处)" % (len(updates), res["updated"]))
if failed:
print("\n失败:")
for fn, why in failed:
print(" %s —— %s" % (fn, why))
print("完成。合同文件在 %s" % C.CONTRACT_DIR)
return 0
if __name__ == "__main__":
sys.exit(main())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment