feat(contracts): 台账新增「合同原件」列（AL），支持上传 / 查阅 / 移除

需求：房租车位费数据表需要支持上传合同留档并可查阅合同，增加合同原件的列。

- 新增 rentos/contracts.py：data/contracts/{id}.{ext} + index.json，
  白名单 pdf/doc/docx/xls/xlsx/png/jpg/gif/webp/bmp，单份 ≤ 20MB；
  save/find/path_of/list_all/remove/parse/dump/describe；clean_id 挡路径穿越
- config：加 CONTRACT_DIR 并纳入 ensure_dirs
- tables：列 AL「合同原件」type=files（排 AJ 之后、最右），一格可挂多份；
  coerce 支持 list → 紧凑 JSON；新增 files_text 供导出 xlsx 输出文件名文本
- server：
  POST /api/contracts（上传，仅管理员）
  GET  /api/contracts（列表）
  GET  /api/contracts/{id}（内联预览；?dl=1 下载；支持 ?token=，便于 <a href> 直接引用）
  DELETE /api/contracts/{id}（仅管理员）
  合同写接口不匹配任何 CAP_ELEC 规则 → 落 _write_cap 兜底 "*" = 仅管理员；
  访客只读仍可查阅（这点由冒烟验证）
- 列定义以代码为准，故加列不需要数据迁移

rentos/config.py

@@ -14,6 +14,7 @@ BATCH_DIR = os.path.join(DATA_DIR, "batches")
 TABLES_DIR = os.path.join(DATA_DIR, "tables")          # 台账类表格存档（汇总表 / 合同收入明细）
 ATTACH_DIR = os.path.join(DATA_DIR, "attachments")     # 电费录入「备注」栏上传的图片附件
 DRAFT_DIR = os.path.join(DATA_DIR, "drafts")           # 电费录入「临时保存」的草稿（按账号一份）
+CONTRACT_DIR = os.path.join(DATA_DIR, "contracts")     # 房租车位费台账「合同原件」（PDF/Word/图片）
 UPLOAD_DIR = os.path.join(BASE, "uploads")
 OUTPUT_DIR = os.path.join(BASE, "output", "elec")       # 电费收费通知书产出
 OUTPUT_RENT_DIR = os.path.join(BASE, "output", "rent")  # 房租车位费收费通知书产出
@@ -142,6 +143,6 @@ PORT = 8899
 
 
 def ensure_dirs():
-    for d in (DATA_DIR, BATCH_DIR, TABLES_DIR, ATTACH_DIR, DRAFT_DIR, UPLOAD_DIR, OUTPUT_DIR,
+    for d in (DATA_DIR, BATCH_DIR, TABLES_DIR, ATTACH_DIR, DRAFT_DIR, CONTRACT_DIR, UPLOAD_DIR, OUTPUT_DIR,
               OUTPUT_RENT_DIR, TABLE_EXPORT_DIR, BATCH_EXPORT_DIR, LOG_DIR, WEB_DIR):
         os.makedirs(d, exist_ok=True)

rentos/contracts.py

+# -*- coding: utf-8 -*-
+"""
+合同原件存档
+============
+房租车位费台账「合同原件」（AL 列）挂的合同扫描件 —— 租赁合同、补充协议、
+车位合同等，可能是 PDF、Word 或手机拍的照片。
+
+存放约定：
+  data/contracts/{id}.{ext}       文件本体
+  data/contracts/index.json       索引 [{id, ext, mime, name, size, at, by}, ...]
+
+设计要点：
+  - id = uuid4().hex（32 位十六进制），进 URL 安全，且不可枚举
+  - 允许 PDF / Word / Excel / 常见图片，单份 ≤ 20MB（合同多有扫描件，比图片宽松）
+  - 前端把 FileReader 读出的 data URL 整串 POST 过来，这里负责剥前缀
+  - 查看接口支持 ?token= 查询参数鉴权，这样 <a href> / <iframe> 能直接引用
+  - 台账里只存「[{id, name}]」这种轻量关联；文件本体独立存放、独立删除
+    （从某一行移除关联不会删文件，别的行还能引用同一份）
+"""
+import base64
+import binascii
+import datetime as dt
+import json
+import os
+import re
+import threading
+import uuid
+
+from . import config as C
+
+MAX_BYTES = 20 * 1024 * 1024         # 单份 20MB（解码后的原始字节）
+
+# MIME → 落盘扩展名
+_MIME_EXT = {
+    "application/pdf": ".pdf",
+    "application/msword": ".doc",
+    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
+    "application/vnd.ms-excel": ".xls",
+    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
+    "image/png": ".png",
+    "image/jpeg": ".jpg",
+    "image/gif": ".gif",
+    "image/webp": ".webp",
+    "image/bmp": ".bmp",
+}
+# 兜底：按原文件名后缀猜 MIME
+_EXT_MIME = {
+    ".pdf": "application/pdf",
+    ".doc": "application/msword",
+    ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+    ".xls": "application/vnd.ms-excel",
+    ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
+    ".jfif": "image/jpeg", ".gif": "image/gif", ".webp": "image/webp",
+    ".bmp": "image/bmp",
+}
+
+_DATA_URL = re.compile(r"^data:([a-z0-9.+-]+/[a-z0-9.+-]+);base64,(.*)$", re.I | re.S)
+_HEX32 = re.compile(r"^[0-9a-f]{32}$")
+
+_LOCK = threading.RLock()
+
+
+# ---------------------------------------------------------------- 索引
+def _index_path():
+    return os.path.join(C.CONTRACT_DIR, "index.json")
+
+
+def _load_index():
+    try:
+        with open(_index_path(), "r", encoding="utf-8") as f:
+            v = json.load(f)
+        return v if isinstance(v, list) else []
+    except (FileNotFoundError, json.JSONDecodeError, OSError):
+        return []
+
+
+def _save_index(items):
+    os.makedirs(C.CONTRACT_DIR, exist_ok=True)
+    p = _index_path()
+    tmp = p + ".tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(items, f, ensure_ascii=False, indent=2)
+    os.replace(tmp, p)
+
+
+def clean_id(cid):
+    """只保留合法 id，否则返回空串（同时挡住路径穿越）。"""
+    s = str(cid or "").strip().lower()
+    return s if _HEX32.match(s) else ""
+
+
+def _safe_name(name):
+    """原始文件名只用于展示：去掉目录部分与危险字符。"""
+    s = str(name or "").replace("\\", "/").split("/")[-1].strip()
+    s = re.sub(r"[\x00-\x1f<>:\"|?*]", "", s)
+    return s[:160]
+
+
+# ---------------------------------------------------------------- 写入
+def _decode(data_b64, filename):
+    """data URL 或裸 base64 → (原始字节, mime)。"""
+    s = str(data_b64 or "").strip()
+    mime = ""
+    m = _DATA_URL.match(s)
+    if m:
+        mime, s = m.group(1).lower(), m.group(2)
+    if mime == "image/jpg":
+        mime = "image/jpeg"
+    if not mime:
+        mime = _EXT_MIME.get(os.path.splitext(_safe_name(filename))[1].lower(), "")
+    if mime not in _MIME_EXT:                 # data URL 带了不认识的类型 → 按扩展名兜底
+        mime = _EXT_MIME.get(os.path.splitext(_safe_name(filename))[1].lower(), mime)
+    if mime not in _MIME_EXT:
+        raise ValueError("只支持 PDF / Word / Excel / 图片（pdf、doc、docx、xls、xlsx、png、jpg、gif、webp、bmp）")
+    s = re.sub(r"\s+", "", s)
+    if not s:
+        raise ValueError("文件内容为空")
+    try:
+        raw = base64.b64decode(s, validate=True)
+    except (binascii.Error, ValueError):
+        raise ValueError("文件内容不是合法的 base64")
+    if not raw:
+        raise ValueError("文件内容为空")
+    if len(raw) > MAX_BYTES:
+        raise ValueError("单个文件不能超过 %d MB" % (MAX_BYTES // 1024 // 1024))
+    return raw, mime
+
+
+def save(data_b64, filename="", username="admin"):
+    """保存一份合同原件，返回它的元信息 dict（含 id）。"""
+    raw, mime = _decode(data_b64, filename)
+    cid = uuid.uuid4().hex
+    ext = _MIME_EXT[mime]
+    with _LOCK:
+        os.makedirs(C.CONTRACT_DIR, exist_ok=True)
+        with open(os.path.join(C.CONTRACT_DIR, cid + ext), "wb") as f:
+            f.write(raw)
+        items = _load_index()
+        meta = {
+            "id": cid,
+            "ext": ext,
+            "mime": mime,
+            "name": _safe_name(filename) or ("合同" + ext),
+            "size": len(raw),
+            "at": dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            "by": username or "",
+        }
+        items.append(meta)
+        _save_index(items)
+    return meta
+
+
+# ---------------------------------------------------------------- 读取
+def find(cid):
+    """按 id 找元信息；找不到返回 None。"""
+    i = clean_id(cid)
+    if not i:
+        return None
+    for it in _load_index():
+        if str(it.get("id")) == i:
+            return it
+    return None
+
+
+def path_of(cid):
+    """按 id 取磁盘路径；文件缺失返回 None。"""
+    it = find(cid)
+    if not it:
+        return None
+    p = os.path.join(C.CONTRACT_DIR, it["id"] + str(it.get("ext") or ""))
+    return p if os.path.isfile(p) else None
+
+
+def list_all(limit=None):
+    items = _load_index()
+    items.sort(key=lambda x: x.get("at") or "", reverse=True)
+    return items[:limit] if limit else items
+
+
+def remove(cid):
+    """删掉文件本体与索引项；返回是否删到了。"""
+    i = clean_id(cid)
+    if not i:
+        return False
+    with _LOCK:
+        items = _load_index()
+        keep = [x for x in items if str(x.get("id")) != i]
+        if len(keep) == len(items):
+            return False
+        p = path_of(i)
+        if p and os.path.isfile(p):
+            try:
+                os.remove(p)
+            except OSError:
+                pass
+        _save_index(keep)
+    return True
+
+
+# ---------------------------------------------------------------- 台账列值
+def parse(value):
+    """台账「合同原件」列值（JSON 文本 / list）→ [{"id","name"}, ...]。
+
+    值不可解析时返回空列表 —— 台账是自由文本，容错优先。
+    """
+    if not value:
+        return []
+    v = value
+    if isinstance(v, str):
+        s = v.strip()
+        if not s:
+            return []
+        try:
+            v = json.loads(s)
+        except ValueError:
+            # 兼容手填的纯 id 串（逗号 / 空白分隔）
+            return [{"id": x, "name": ""} for x in re.split(r"[,\s]+", s) if clean_id(x)]
+    if not isinstance(v, list):
+        return []
+    out, seen = [], set()
+    for it in v:
+        if isinstance(it, str):
+            it = {"id": it, "name": ""}
+        if not isinstance(it, dict):
+            continue
+        cid = clean_id(it.get("id"))
+        if not cid or cid in seen:
+            continue
+        seen.add(cid)
+        meta = find(cid) or {}
+        out.append({"id": cid, "name": str(it.get("name") or meta.get("name") or "")})
+    return out
+
+
+def dump(items):
+    """[{"id","name"}] → 台账单元格存的紧凑 JSON 文本（空列表 → ""）。"""
+    clean = []
+    for it in items or []:
+        cid = clean_id((it or {}).get("id"))
+        if not cid:
+            continue
+        meta = find(cid) or {}
+        clean.append({"id": cid, "name": str((it or {}).get("name") or meta.get("name") or "")})
+    return json.dumps(clean, ensure_ascii=False) if clean else ""
+
+
+def describe(value):
+    """给前端用：台账列值 → 带 url / 大小 / 是否还在的明细列表。"""
+    out = []
+    for it in parse(value):
+        meta = find(it["id"])
+        row = {
+            "id": it["id"],
+            "name": it.get("name") or (meta or {}).get("name") or "合同原件",
+            "url": "/api/contracts/" + it["id"],
+            "exists": bool(meta),
+        }
+        if meta:
+            row.update({"size": meta.get("size") or 0, "mime": meta.get("mime") or "",
+                        "at": meta.get("at") or "", "ext": meta.get("ext") or ""})
+        out.append(row)
+    return out

rentos/server.py

@@ -54,6 +54,12 @@ Python 标准库 http.server，零外部依赖。启动：
     GET    /api/attachments               附件列表
     GET    /api/attachments/{id}          在线查看图片（支持 ?token=，便于 <img src>）
     DELETE /api/attachments/{id}          删除附件
+
+    合同原件（房租车位费台账「合同原件」列，仅管理员可写）
+    POST   /api/contracts                 上传合同（base64 / data URL）→ 合同 id
+    GET    /api/contracts                 合同列表
+    GET    /api/contracts/{id}            查阅合同（内联预览；?dl=1 下载；支持 ?token=）
+    DELETE /api/contracts/{id}            删除合同文件本体
     GET    /api/scheduler                 定时器状态（含两张账单各自的计划）
     POST   /api/scheduler/fire            立即触发一次自动发送（可指定 plan）
     GET    /api/rentfree                  免租期窗口 / 当前处于免租期的单元
@@ -88,6 +94,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from . import attachments
 from . import auth
 from . import config as C
+from . import contracts
 from . import directory
 from . import masters
 from . import plans
@@ -229,7 +236,13 @@ class Handler(BaseHTTPRequestHandler):
 
     _CTYPES = {
         ".pdf": "application/pdf",
+        ".doc": "application/msword",
+        ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+        ".xls": "application/vnd.ms-excel",
         ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        ".png": "image/png",
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
         ".zip": "application/zip",
     }
 
@@ -394,6 +407,27 @@ class Handler(BaseHTTPRequestHandler):
                                     meta.get("name") or (meta["id"] + str(meta.get("ext") or "")),
                                     meta.get("mime") or "application/octet-stream")
 
+            # ---- 合同原件（房租车位费台账「合同原件」列）
+            if path == "/api/contracts":
+                q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
+                limit = int((q.get("limit") or ["500"])[0])
+                return self._send(200, {"ok": True,
+                                        "contracts": contracts.list_all(limit)})
+
+            m = re.match(r"^/api/contracts/([A-Za-z0-9]+)$", path)
+            if m:
+                q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
+                meta = contracts.find(m.group(1))
+                p = contracts.path_of(m.group(1))
+                if not meta or not p:
+                    return self._send(404, {"ok": False, "error": "合同文件不存在"})
+                name = meta.get("name") or (meta["id"] + str(meta.get("ext") or ""))
+                ctype = meta.get("mime") or self._CTYPES.get(
+                    str(meta.get("ext") or "").lower(), "application/octet-stream")
+                if (q.get("dl") or ["0"])[0] in ("1", "true", "yes"):
+                    return self._download(p, name)          # 下载（Word 等本地打不开的格式）
+                return self._inline(p, name, ctype)         # PDF / 图片直接在线预览
+
             # ---- 房租车位费台账
             if path == "/api/tables":
                 return self._send(200, {"ok": True, "tables": tables.list_tables()})
@@ -775,6 +809,17 @@ class Handler(BaseHTTPRequestHandler):
             if m and method == "DELETE":
                 return self._send(200, {"ok": attachments.remove(m.group(1))})
 
+            # ---- 合同原件（房租车位费台账「合同原件」列，仅管理员）
+            if path == "/api/contracts" and method == "POST":
+                meta = dict(contracts.save(body.get("data"), body.get("filename"),
+                                           username=user["username"]))
+                meta["url"] = "/api/contracts/" + meta["id"]
+                return self._send(200, {"ok": True, "contract": meta})
+
+            m = re.match(r"^/api/contracts/([A-Za-z0-9]+)$", path)
+            if m and method == "DELETE":
+                return self._send(200, {"ok": contracts.remove(m.group(1))})
+
             m = re.match(r"^/api/batches/([^/]+)/rerender$", path)
             if m and method == "POST":
                 b = service.rerender(m.group(1),

rentos/tables.py

@@ -37,7 +37,9 @@ from . import config as C
 _LOCK = threading.RLock()
 
 # 允许的列类型
-TYPES = ("text", "int", "num", "money", "date", "pct")
+#   files = 附件列（目前只有「合同原件」）：值是 [{id, name}] 的紧凑 JSON，
+#           本体由 rentos/contracts.py 管，前端渲染成可点击的查阅按钮
+TYPES = ("text", "int", "num", "money", "date", "pct", "files")
 
 RENT_KEY = "rent_parking"
 
@@ -91,6 +93,12 @@ _RENT_PARKING_COLS = [
     #      出单口径仍看「账单起始日」(AC)；两列在「出单推进账期」时一起往后走，
     #      需要提前/延后推送时可直接改本列（租户拖账期也不用动出单逻辑）。
     _c("AJ", "下一账单日", "推送", "date", 100),
+    # ---- 合同原件（2026-10-15 新增）
+    #      单元格存紧凑 JSON：[{"id":"…","name":"302租赁合同.pdf"}, …]，可挂多份
+    #      （租赁合同 + 补充协议 + 车位合同）。
+    #      文件本体在 data/contracts/（rentos/contracts.py），从某行移除关联不会删文件。
+    #      键用 AL（不占用字母顺序），显示位置在最后，便于浏览时一眼看到有无合同。
+    _c("AL", "合同原件", "", "files", 168),
 ]
 
 # 出单所需的列（供出单模块引用，避免散落魔法字符串）
@@ -163,8 +171,55 @@ def save_table(key, table):
 
 
 # ================================================================ 取值 / 规范化
+def _files_value(value):
+    """「合同原件」列落库前的规整：list → 紧凑 JSON 文本；字符串原样；空 → None。"""
+    if value is None:
+        return None
+    if isinstance(value, str):
+        s = value.strip()
+        return None if (not s or s == "[]") else s
+    if isinstance(value, list):
+        items = []
+        for it in value:
+            if isinstance(it, str):
+                it = {"id": it}
+            if not isinstance(it, dict):
+                continue
+            i = str(it.get("id") or "").strip()
+            if not i:
+                continue
+            items.append({"id": i, "name": str(it.get("name") or "")})
+        return json.dumps(items, ensure_ascii=False) if items else None
+    return None
+
+
+def files_text(value):
+    """「合同原件」列值 → 便于阅读 / 导出的文件名文本（一行一份）。"""
+    if not value:
+        return ""
+    v = value
+    if isinstance(v, str):
+        try:
+            v = json.loads(v)
+        except ValueError:
+            return v
+    if not isinstance(v, list):
+        return str(value)
+    names = []
+    for it in v:
+        if isinstance(it, str):
+            names.append(it)
+        elif isinstance(it, dict):
+            n = str(it.get("name") or it.get("id") or "").strip()
+            if n:
+                names.append(n)
+    return "\n".join(names)
+
+
 def coerce(value, ctype):
     """前端提交值 → 存档值。空串一律存 None。"""
+    if ctype == "files":
+        return _files_value(value)
     if value is None:
         return None
     if isinstance(value, str):
@@ -461,10 +516,11 @@ def export_xlsx(key, out_path=None):
     r = 5
     for row in t["rows"]:
         for i, c in enumerate(cols, start=1):
-            cell = ws.cell(r, i, row.get(c["id"]))
+            val = files_text(row.get(c["id"])) if c["type"] == "files" else row.get(c["id"])
+            cell = ws.cell(r, i, val)
             cell.font = F_BODY
             cell.border = B
-            cell.alignment = Alignment(vertical="center",
+            cell.alignment = Alignment(vertical="center", wrap_text=(c["type"] == "files"),
                                        horizontal="right" if c["type"] in
                                        ("int", "num", "money", "pct") else "left")
             if c["type"] == "money":