Commit a39ca9ca authored by wenwen.tang's avatar wenwen.tang 😕

feat(收款+权限): 收款日期同步台账最近收款月份;新增只读访客账户

1) 收款日期 → 台账账期同步
- set_receipt 登记/修改收款日期时,把该单据覆盖的台账行「最近收款月份」(AB)
  同步为【收款日期所在月】(口径经确认);清空收款日期则退回同步前的值
- 批次记 receipt_sync 记录,删除批次时一并回滚「最近收款月份」
- 批次明细快照的 last_paid 同步跟进,页面不会出现两个口径
- 修:不能拿 service._d()(失败回落"今天")算月份,改用 rent_parser._d()

2) 只读访客账户
- auth.py 支持多账号 + 角色(admin 可读写 / viewer 只读),首次启动播种
  admin(1989@rent05) 与 guest(guest@2026);未知角色一律降级 viewer;
  主账号不可删;访客被删后不再自动复活
- server.py 增加写闸门:除 /api/logout 外,viewer 的 POST/PUT/DELETE 一律 403
- 新增账号管理接口 GET/POST /api/accounts、DELETE /api/accounts/{u}、
  POST /api/accounts/{u}/password(仅管理员)
- 前端按角色隐藏写入口 + api() 兜底拦截;设置页新增「账号管理」卡片

测试
- tools/test_accounts.py(新增 34 项,含真 HTTP 写闸门实测)
- tools/test_round5.py 补 11 项收款同步用例;全量 191 项通过
- tools/ui_verify7.py(新增,自带沙箱数据目录 + 真浏览器)59/59 通过
parent deac4306
# -*- coding: utf-8 -*-
"""
登录校验与会话管理
==================
账密校验:账号 admin / 密码 1989@rent05(首次启动自动写入加盐哈希,不落明文)
会话:登录成功下发 token(secrets.token_urlsafe),存 data/sessions.json,默认 12 小时过期。
登录校验、多账号与角色(会话管理)
==================================
账号:data/config.json 的 auth.accounts 列表
[{"username","salt","password_hash","role","note","created_at"}]
role = "admin"(可读写) | "viewer"(只读访客,服务端一律拒绝写操作)
默认账号(首次启动自动播种加盐哈希,不落明文)
- admin / 1989@rent05 管理员,可读写
- guest / guest@2026 只读访客,只能浏览
兼容:旧版 auth 里只有 username / salt / password_hash 三个字段时,
首次 ensure_default_account() 会把它补成 accounts 里的管理员条目;
主账号(auth.username)始终是管理员,且不允许被删除。
会话:登录成功下发 token(secrets.token_urlsafe),存 data/sessions.json,
默认 12 小时过期;会话里同时记下 username 与 role。
"""
import datetime as dt
import hashlib
......@@ -15,42 +27,206 @@ from . import store
SESSION_TTL_HOURS = 12
ROLE_ADMIN = "admin"
ROLE_VIEWER = "viewer"
ROLES = (ROLE_ADMIN, ROLE_VIEWER)
DEFAULT_ADMIN = ("admin", "1989@rent05", "系统管理员")
DEFAULT_GUEST = ("guest", "guest@2026", "只读访客")
# ---------------------------------------------------------------- 密码
def make_hash(password, salt):
return hashlib.sha256((salt + password).encode("utf-8")).hexdigest()
def new_salt():
return secrets.token_hex(8)
def _account(username, password, role, note=""):
salt = new_salt()
return {
"username": str(username or "").strip(),
"salt": salt,
"password_hash": make_hash(str(password or ""), salt),
"role": role,
"note": note,
"created_at": dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
# ---------------------------------------------------------------- 账号
def ensure_default_account():
"""首次启动:若未设置密码,写入默认账号 admin / 1989@rent05。"""
"""首次启动:把主账号与访客账号补齐(幂等,只在有变化时落盘)。"""
cfg = store.load_config()
auth = cfg.setdefault("auth", {})
changed = False
# 旧结构:只有 username / salt / password_hash
if not auth.get("password_hash"):
auth["username"] = auth.get("username") or "admin"
auth["salt"] = auth.get("salt") or secrets.token_hex(8)
auth["password_hash"] = make_hash("1989@rent05", auth["salt"])
auth["username"] = auth.get("username") or DEFAULT_ADMIN[0]
auth["salt"] = auth.get("salt") or new_salt()
auth["password_hash"] = make_hash(DEFAULT_ADMIN[1], auth["salt"])
changed = True
accounts = auth.get("accounts")
if not isinstance(accounts, list):
accounts = []
changed = True
def _has(name):
return any(str(a.get("username") or "") == name for a in accounts)
admin_name = str(auth.get("username") or DEFAULT_ADMIN[0])
# 主账号 → accounts(role 固定为管理员)
row = next((a for a in accounts if str(a.get("username") or "") == admin_name), None)
if row is None:
accounts.insert(0, {
"username": admin_name,
"salt": auth.get("salt") or new_salt(),
"password_hash": auth.get("password_hash") or make_hash(DEFAULT_ADMIN[1], new_salt()),
"role": ROLE_ADMIN,
"note": DEFAULT_ADMIN[2],
"created_at": dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
})
changed = True
elif normalize_role(row.get("role")) != ROLE_ADMIN:
row["role"] = ROLE_ADMIN
changed = True
# 访客账号:只在首次初始化时播种一次;
# 之后即便被管理员删掉也不会自动复活(靠 guest_seeded 标记)。
if not auth.get("guest_seeded"):
auth["guest_seeded"] = True
changed = True
if not _has(DEFAULT_GUEST[0]):
accounts.append(_account(DEFAULT_GUEST[0], DEFAULT_GUEST[1],
ROLE_VIEWER, DEFAULT_GUEST[2]))
auth["accounts"] = accounts
if changed:
store.save_config(cfg)
return auth
def normalize_role(role):
"""未知 / 缺省角色一律按只读处理,避免误放行。"""
r = str(role or "").strip().lower()
return r if r in ROLES else ROLE_VIEWER
def accounts():
ensure_default_account()
cfg = store.load_config()
out = (cfg.get("auth") or {}).get("accounts")
return [dict(a) for a in out] if isinstance(out, list) else []
def admin_username():
cfg = store.load_config()
return str((cfg.get("auth") or {}).get("username") or DEFAULT_ADMIN[0])
def find(username):
u = str(username or "").strip()
for a in accounts():
if str(a.get("username") or "") == u:
return a
return None
def authenticate(username, password):
"""账密校验 → 账号 dict(含 role);失败返回 None。"""
u = str(username or "").strip()
if not u:
return None
for a in accounts():
if not hmac.compare_digest(u, str(a.get("username") or "")):
continue
h = make_hash(str(password or ""), a.get("salt") or "")
if hmac.compare_digest(h, str(a.get("password_hash") or "")):
a["role"] = normalize_role(a.get("role"))
return a
return None
def verify(username, password):
auth = ensure_default_account()
if not hmac.compare_digest(str(username or ""), str(auth.get("username") or "")):
return False
h = make_hash(str(password or ""), auth.get("salt") or "")
return hmac.compare_digest(h, auth.get("password_hash") or "")
"""兼容旧调用:只判断是否通过。"""
return authenticate(username, password) is not None
def public_accounts():
"""下发给页面的账号清单(不含任何哈希 / 盐)。"""
main = admin_username()
return [{
"username": a.get("username"),
"role": normalize_role(a.get("role")),
"note": a.get("note") or "",
"created_at": a.get("created_at") or "",
"primary": str(a.get("username")) == main,
} for a in accounts()]
def set_password(new_password, username=None):
"""改密码:username 为空时改主账号。同步旧字段与 accounts。"""
ensure_default_account()
cfg = store.load_config()
auth = cfg.setdefault("auth", {})
target = str(username or auth.get("username") or DEFAULT_ADMIN[0])
salt = new_salt()
h = make_hash(new_password, salt)
hit = False
for a in auth.get("accounts") or []:
if str(a.get("username") or "") == target:
a["salt"], a["password_hash"] = salt, h
hit = True
if not hit:
raise ValueError("账号不存在:%s" % target)
if target == str(auth.get("username") or ""):
auth["salt"], auth["password_hash"] = salt, h
store.save_config(cfg)
return target
def add_account(username, password, role=ROLE_VIEWER, note=""):
"""新增账号(已存在则报错)。"""
ensure_default_account()
u = str(username or "").strip()
if not u:
raise ValueError("账号不能为空")
if len(str(password or "")) < 6:
raise ValueError("密码至少 6 位")
if find(u):
raise ValueError("账号已存在:%s" % u)
cfg = store.load_config()
auth = cfg.setdefault("auth", {})
auth.setdefault("accounts", []).append(
_account(u, password, normalize_role(role), note))
store.save_config(cfg)
return find(u)
def set_password(new_password):
def remove_account(username):
"""删除账号。主账号不可删。"""
ensure_default_account()
u = str(username or "").strip()
if u == admin_username():
raise ValueError("主账号 %s 不可删除" % u)
cfg = store.load_config()
auth = cfg.setdefault("auth", {})
auth["salt"] = secrets.token_hex(8)
auth["password_hash"] = make_hash(new_password, auth["salt"])
rows = auth.get("accounts") or []
kept = [a for a in rows if str(a.get("username") or "") != u]
if len(kept) == len(rows):
raise ValueError("账号不存在:%s" % u)
auth["accounts"] = kept
store.save_config(cfg)
return True
# ---------------------------------------------------------------- 会话
def create_session(username, ttl_hours=SESSION_TTL_HOURS):
def create_session(username, role=ROLE_ADMIN, ttl_hours=SESSION_TTL_HOURS):
sessions = store.load_sessions()
now = dt.datetime.now()
# 顺手清理过期会话
......@@ -59,6 +235,7 @@ def create_session(username, ttl_hours=SESSION_TTL_HOURS):
token = secrets.token_urlsafe(32)
sessions[token] = {
"username": username,
"role": normalize_role(role),
"created_at": now.strftime("%Y-%m-%d %H:%M:%S"),
"expire_at": (now + dt.timedelta(hours=ttl_hours)).strftime("%Y-%m-%d %H:%M:%S"),
}
......@@ -78,6 +255,8 @@ def check_session(token):
sessions.pop(token, None)
store.save_sessions(sessions)
return None
# 老会话没有 role 字段 → 当年只有管理员能登录,按管理员处理
s["role"] = normalize_role(s["role"]) if s.get("role") else ROLE_ADMIN
return s
......
......@@ -47,6 +47,12 @@ DEFAULT_CONFIG = {
# sha256(salt + password),默认密码 1989@rent05
"salt": "rentos-2026",
"password_hash": "",
# 多账号 + 角色。role: "admin" 可读写 | "viewer" 只读访客
# [{"username","salt","password_hash","role","note","created_at"}]
# 首次启动会播种 admin(1989@rent05)与 guest(guest@2026,只读)
"accounts": [],
# 访客账号只播种一次;管理员删掉后不再自动复活
"guest_seeded": False,
},
"notify": {
# 定时发送总开关;关闭后两张账单都不自动发送,只能手动发
......
......@@ -7,9 +7,13 @@ Python 标准库 http.server,零外部依赖。启动:
python -m rentos.server --port 8899 --no-scheduler
路由总览(全部 JSON,除静态与文件下载)
POST /api/login 账密登录 → token
POST /api/login 账密登录 → token(含 role:admin 可读写 / viewer 只读)
POST /api/logout 退出
GET /api/session 当前会话
GET /api/session 当前会话(含 role / can_write)
GET /api/accounts 账号清单(仅管理员)
POST /api/accounts 新增账号(仅管理员)
DELETE /api/accounts/{username} 删除账号(仅管理员;主账号不可删)
POST /api/accounts/{username}/password 重置指定账号密码(仅管理员)
GET /api/overview 看板
GET /api/config 读配置
PUT /api/config 改配置(通知时间 / 电费参数 / 企微设置)
......@@ -25,6 +29,7 @@ Python 标准库 http.server,零外部依赖。启动:
DELETE /api/batches/{id} 删除批次(body.rollback=true 时回滚账期推进)
POST /api/batches/{id}/rerender 按最新配置重出单据
POST /api/batches/{id}/receipt 登记收款日期 / 是否开票(勾上即记开票日期)
房租批次会同步台账「最近收款月份」
POST /api/batches/{id}/send 手动发送(可指定单据序号)
GET /api/batches/{id}/export 导出批次数据(xlsx:单据 + 明细 + 汇总)
GET /api/batches/{id}/file/{i} 下载第 i 份(PDF)
......@@ -133,6 +138,13 @@ class Handler(BaseHTTPRequestHandler):
def _user(self):
return auth.check_session(self._token())
@staticmethod
def _role(user):
"""会话角色。缺省按管理员(老会话),未知角色按只读处理。"""
if not user:
return ""
return auth.normalize_role(user.get("role") or auth.ROLE_ADMIN)
def _guard(self):
u = self._user()
if not u:
......@@ -140,6 +152,14 @@ class Handler(BaseHTTPRequestHandler):
return None
return u
def _guard_write(self, user):
"""写操作闸门:只读访客一律拒绝(服务端硬拦截,不依赖前端隐藏按钮)。"""
if self._role(user) == auth.ROLE_ADMIN:
return True
self._send(403, {"ok": False, "readonly": True,
"error": "只读访客账号:仅可浏览,无修改权限"})
return False
_CTYPES = {
".pdf": "application/pdf",
".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
......@@ -213,7 +233,9 @@ class Handler(BaseHTTPRequestHandler):
if path == "/api/session":
u = self._user()
return self._send(200, {"ok": True, "logged_in": bool(u),
"username": (u or {}).get("username", "")})
"username": (u or {}).get("username", ""),
"role": self._role(u),
"can_write": self._role(u) == auth.ROLE_ADMIN})
# ---- 需登录
user = self._guard()
......@@ -228,6 +250,13 @@ class Handler(BaseHTTPRequestHandler):
cfg.pop("auth", None) # 不下发账号信息
return self._send(200, {"ok": True, "config": cfg})
if path == "/api/accounts":
if self._role(user) != auth.ROLE_ADMIN:
return self._send(403, {"ok": False, "error": "仅管理员可查看账号"})
return self._send(200, {"ok": True, "accounts": auth.public_accounts(),
"me": user.get("username"),
"my_role": self._role(user)})
if path == "/api/owners":
return self._send(200, {"ok": True,
"owners": service.owners_config(),
......@@ -400,10 +429,14 @@ class Handler(BaseHTTPRequestHandler):
if path == "/api/login" and method == "POST":
u = str(body.get("username") or "").strip()
p = str(body.get("password") or "")
if not auth.verify(u, p):
acc = auth.authenticate(u, p)
if not acc:
return self._send(401, {"ok": False, "error": "账号或密码不正确"})
token = auth.create_session(u)
return self._send(200, {"ok": True, "token": token, "username": u})
role = acc.get("role") or auth.ROLE_ADMIN
token = auth.create_session(acc["username"], role)
return self._send(200, {"ok": True, "token": token,
"username": acc["username"], "role": role,
"can_write": role == auth.ROLE_ADMIN})
user = self._guard()
if not user:
......@@ -413,6 +446,10 @@ class Handler(BaseHTTPRequestHandler):
auth.drop_session(self._token())
return self._send(200, {"ok": True})
# ---- 只读访客:除退出登录外,所有写操作一律拒绝
if not self._guard_write(user):
return
if path == "/api/password" and method == "POST":
newp = str(body.get("new_password") or "")
if len(newp) < 6:
......@@ -420,6 +457,32 @@ class Handler(BaseHTTPRequestHandler):
auth.set_password(newp)
return self._send(200, {"ok": True})
# ================================================ 账号管理(仅管理员)
if path == "/api/accounts" and method == "POST":
row = auth.add_account(str(body.get("username") or ""),
str(body.get("password") or ""),
role=str(body.get("role") or auth.ROLE_VIEWER),
note=body.get("note") or "")
return self._send(200, {"ok": True,
"account": {"username": row["username"],
"role": row["role"],
"note": row.get("note") or ""},
"accounts": auth.public_accounts()})
m = re.match(r"^/api/accounts/([^/]+)/password$", path)
if m and method == "POST":
newp = str(body.get("new_password") or "")
if len(newp) < 6:
return self._send(400, {"ok": False, "error": "新密码至少 6 位"})
target = auth.set_password(newp, username=m.group(1))
return self._send(200, {"ok": True, "username": target})
m = re.match(r"^/api/accounts/([^/]+)$", path)
if m and method == "DELETE":
auth.remove_account(m.group(1))
return self._send(200, {"ok": True, "removed": m.group(1),
"accounts": auth.public_accounts()})
if path == "/api/config" and method == "PUT":
allow = ("notify", "wecom", "elec", "remit")
partial = {k: v for k, v in body.items() if k in allow}
......@@ -540,6 +603,7 @@ class Handler(BaseHTTPRequestHandler):
rollback = None
if b and (b.get("kind") or "elec") == service.RENT_KEY and body.get("rollback"):
rollback = service.rollback_rent_advance(b)
service.rollback_receipt_sync(b) # 收款登记同步过的「最近收款月份」一并退回
ok = store.delete_batch(bid)
return self._send(200, {"ok": ok, "rollback": rollback})
......@@ -563,12 +627,13 @@ class Handler(BaseHTTPRequestHandler):
m = re.match(r"^/api/batches/([^/]+)/receipt$", path)
if m and method == "POST":
b = service.set_receipt(m.group(1), body.get("index"),
b, sync = service.set_receipt(m.group(1), body.get("index"),
paid_at=body.get("paid_at"),
invoiced=body.get("invoiced"),
invoiced_at=body.get("invoiced_at"))
return self._send(200, {"ok": True, "batch": b,
"receipt": service.receipt_summary(b)})
"receipt": service.receipt_summary(b),
"sync": sync})
m = re.match(r"^/api/batches/([^/]+)/send$", path)
if m and method == "POST":
......
......@@ -522,12 +522,106 @@ def prune_output(batch_id, keep_files, kind="elec"):
# ---------------------------------------------------------------- 收款 / 开票登记
def _month_of(s):
"""日期 → 「YYYY-MM」;空值 / 解析不了返回空串。
注意用 rent_parser._d:它解析失败返回 None;
本模块的 _d 为出单容错会回落「今天」,不能用于此处。
"""
d = RP._d(s)
return d.strftime("%Y-%m") if d else ""
def _sync_records(batch):
rec = batch.get("receipt_sync")
return rec if isinstance(rec, list) else []
def sync_receipt_month(batch, index, doc):
"""登记收款后,把该单据覆盖的台账行「最近收款月份」(AB) 同步为收款月份。
口径(2026-10-13 与巴特菲确认):取**收款日期所在月**("YYYY-MM")。
清空收款日期时回退 —— 仅当台账当前值仍等于上次同步写入的值才退回,
期间被人工改过的行不动(与「删除批次回滚账期」同一保守策略)。
批次里留一份 receipt_sync 记录({doc,row_id,from,to}),
供清空收款日期、删除批次时逐步回退。
返回 {"updated","reverted","month","rows":[{"row_id","from","to"}]}。
"""
col = tables.BILL_COLS["last_paid"]
rid_list = [str(r) for r in (doc.get("row_ids") or []) if str(r).strip()]
target = _month_of(doc.get("paid_at"))
keep = [r for r in _sync_records(batch) if r.get("doc") != index]
old = {r.get("row_id"): r for r in _sync_records(batch) if r.get("doc") == index}
table = tables.load_table(RENT_KEY)
cur = {r.get("_id"): r.get(col) for r in (table or {}).get("rows") or []}
ups, rows_out, recs = [], [], list(keep)
detail_up = {}
reverted = 0
for rid in rid_list:
if rid not in cur:
continue
before = cur.get(rid)
prev = old.get(rid)
if target:
if str(before or "")[:7] == target:
# 已是目标月份:保留原有 from,方便后续回退
if prev:
recs.append(prev)
continue
frm = prev.get("from") if prev else before
ups.append((rid, col, target))
recs.append({"doc": index, "row_id": rid, "from": frm, "to": target})
rows_out.append({"row_id": rid, "from": frm, "to": target})
detail_up[rid] = target
elif prev and str(before or "")[:7] == str(prev.get("to") or ""):
ups.append((rid, col, prev.get("from")))
rows_out.append({"row_id": rid, "from": prev.get("to"), "to": prev.get("from")})
detail_up[rid] = prev.get("from")
reverted += 1
if ups:
tables.update_cells(RENT_KEY, ups, who="登记收款同步最近收款月份")
batch["receipt_sync"] = recs
# 批次明细快照里的「最近收款月份」一起跟进,避免页面出现两个口径
for d in batch.get("details") or []:
if d.get("row_id") in detail_up:
d["last_paid"] = detail_up[d["row_id"]] or ""
return {"updated": len(ups) - reverted, "reverted": reverted,
"month": target, "rows": rows_out}
def rollback_receipt_sync(batch):
"""把该批次登记收款时同步过的「最近收款月份」退回原值(删除批次时调用)。"""
recs = _sync_records(batch)
if not recs:
return {"reverted": 0, "skipped": 0}
col = tables.BILL_COLS["last_paid"]
table = tables.load_table(RENT_KEY)
cur = {r.get("_id"): r.get(col) for r in (table or {}).get("rows") or []}
ups = []
for r in recs:
rid = r.get("row_id")
if rid in cur and str(cur.get(rid) or "")[:7] == str(r.get("to") or ""):
ups.append((rid, col, r.get("from")))
res = tables.update_cells(RENT_KEY, ups, who="删除批次回滚收款月份")
return {"reverted": res.get("updated", 0),
"skipped": len(recs) - res.get("updated", 0)}
def set_receipt(batch_id, index, paid_at=None, invoiced=None, invoiced_at=None):
"""登记某一份单据的「收款日期」与「是否开票」。
paid_at "" 或 YYYY-MM-DD(空串 = 清空,表示还没收到钱)
invoiced True/False;勾上时若没给 invoiced_at 就默认记今天
返回最新批次;越界的 index 抛 ValueError。
越界的 index 抛 ValueError。
房租车位费批次还会把台账「最近收款月份」同步成收款日期所在月。
返回 (batch, sync) —— sync 给出这次同步改了几行。
"""
batch = store.load_batch(batch_id)
if not batch:
......@@ -555,9 +649,17 @@ def set_receipt(batch_id, index, paid_at=None, invoiced=None, invoiced_at=None):
elif invoiced_at is not None and doc.get("invoiced"):
doc["invoiced_at"] = str(invoiced_at or "").strip()[:10]
# ---- 同步台账「最近收款月份」(仅房租车位费批次;失败不阻塞登记)
sync = {"updated": 0, "reverted": 0, "month": "", "rows": [], "error": ""}
if paid_at is not None and (batch.get("kind") or "elec") == RENT_KEY:
try:
sync = sync_receipt_month(batch, i, doc)
except Exception as e: # noqa: BLE001
sync["error"] = str(e)
batch["receipt_updated_at"] = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
store.save_batch(batch)
return batch
return batch, sync
def receipt_summary(batch):
......
This diff is collapsed.
......@@ -20,6 +20,7 @@
运行:python -m unittest discover -s tools -p "test_*.py"
"""
import datetime as dt
import json
import os
import shutil
import sys
......@@ -82,7 +83,7 @@ class TestReceipt(FakeBatchBackend):
def test_set_paid_at(self):
b = self._batch()
self.assertEqual(len(b["documents"]), 2)
b2 = service.set_receipt(b["batch_id"], 1, paid_at="2026-10-15")
b2, sync = service.set_receipt(b["batch_id"], 1, paid_at="2026-10-15")
self.assertEqual(b2["documents"][1]["paid_at"], "2026-10-15")
s = service.receipt_summary(b2)
self.assertEqual(s["doc_count"], 2)
......@@ -92,10 +93,15 @@ class TestReceipt(FakeBatchBackend):
round(b2["documents"][1]["total_amount"], 2), places=2)
self.assertAlmostEqual(s["unpaid_amount"],
round(b2["documents"][0]["total_amount"], 2), places=2)
# 收款日期 → 台账「最近收款月份」= 收款日期所在月
self.assertEqual(sync["month"], "2026-10")
self.assertEqual(sync["updated"], 1)
rid = b2["documents"][1]["row_ids"][0]
self.assertEqual(self.row(rid)["AB"], "2026-10")
def test_invoice_checked_records_today(self):
b = self._batch()
b2 = service.set_receipt(b["batch_id"], 0, invoiced=True)
b2, _ = service.set_receipt(b["batch_id"], 0, invoiced=True)
d = b2["documents"][0]
self.assertTrue(d["invoiced"])
self.assertEqual(d["invoiced_at"], dt.date.today().strftime("%Y-%m-%d"))
......@@ -105,28 +111,28 @@ class TestReceipt(FakeBatchBackend):
def test_invoice_with_explicit_date(self):
b = self._batch()
b2 = service.set_receipt(b["batch_id"], 0, invoiced=True, invoiced_at="2026-10-20")
b2, _ = service.set_receipt(b["batch_id"], 0, invoiced=True, invoiced_at="2026-10-20")
self.assertEqual(b2["documents"][0]["invoiced_at"], "2026-10-20")
def test_uncheck_clears_invoice_date(self):
b = self._batch()
b = service.set_receipt(b["batch_id"], 0, invoiced=True, invoiced_at="2026-10-20")
b = service.set_receipt(b["batch_id"], 0, invoiced=False)
b, _ = service.set_receipt(b["batch_id"], 0, invoiced=True, invoiced_at="2026-10-20")
b, _ = service.set_receipt(b["batch_id"], 0, invoiced=False)
self.assertFalse(b["documents"][0]["invoiced"])
self.assertEqual(b["documents"][0]["invoiced_at"], "")
self.assertEqual(service.receipt_summary(b)["invoiced_count"], 0)
def test_clear_paid_at(self):
b = self._batch()
b = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-15")
b = service.set_receipt(b["batch_id"], 0, paid_at="")
b, _ = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-15")
b, _ = service.set_receipt(b["batch_id"], 0, paid_at="")
self.assertEqual(b["documents"][0]["paid_at"], "")
self.assertEqual(service.receipt_summary(b)["paid_count"], 0)
self.assertEqual(service.receipt_summary(b)["last_paid_at"], "")
def test_receipt_updated_at_stamped(self):
b = self._batch()
b2 = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-15")
b2, _ = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-15")
self.assertTrue(b2.get("receipt_updated_at"))
def test_index_out_of_range(self):
......@@ -146,6 +152,104 @@ class TestReceipt(FakeBatchBackend):
self.assertEqual(s["last_paid_at"], "")
# ================================================================ 1b. 收款 → 台账账期同步
class TestReceiptLedgerSync(FakeBatchBackend):
"""登记收款日期 → 台账「最近收款月份」(AB) 同步为收款日期所在月。"""
def _batch(self):
return service.ingest_rent(row_ids=["r3", "r6"], on_date="2026-10-09",
notice_date="2026-10-09", advance=False)
def test_month_is_taken_from_paid_date(self):
b = self._batch()
rid = b["documents"][0]["row_ids"][0]
self.assertEqual(self.row(rid)["AB"], None) # 出单前台账没有收款月份
_, sync = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
self.assertEqual(sync["month"], "2026-10")
self.assertEqual(self.row(rid)["AB"], "2026-10")
self.assertEqual(sync["updated"], 1)
def test_month_follows_changed_paid_date(self):
b = self._batch()
rid = b["documents"][0]["row_ids"][0]
service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
self.assertEqual(self.row(rid)["AB"], "2026-10")
service.set_receipt(b["batch_id"], 0, paid_at="2026-11-05")
self.assertEqual(self.row(rid)["AB"], "2026-11")
def test_clearing_paid_date_reverts_ledger(self):
b = self._batch()
rid = b["documents"][0]["row_ids"][0]
self.row(rid)["AB"] = "2026-09" # 台账原值
service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
self.assertEqual(self.row(rid)["AB"], "2026-10")
_, sync = service.set_receipt(b["batch_id"], 0, paid_at="")
self.assertEqual(sync["reverted"], 1)
self.assertEqual(self.row(rid)["AB"], "2026-09") # 退回同步前的值
def test_manual_edit_is_not_clobbered_on_clear(self):
"""期间被人工改过的行,清空收款日期时不动它。"""
b = self._batch()
rid = b["documents"][0]["row_ids"][0]
service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
self.row(rid)["AB"] = "2026-12" # 人工改成别的值
_, sync = service.set_receipt(b["batch_id"], 0, paid_at="")
self.assertEqual(sync["reverted"], 0)
self.assertEqual(self.row(rid)["AB"], "2026-12")
def test_all_rows_of_document_are_synced(self):
b = self._batch()
doc = b["documents"][0]
_, sync = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
for rid in doc["row_ids"]:
self.assertEqual(self.row(rid)["AB"], "2026-10")
self.assertEqual(sync["updated"], len(doc["row_ids"]))
def test_empty_paid_at_does_not_write(self):
b = self._batch()
before = [json.loads(json.dumps(r)) for r in self.ledger]
service.set_receipt(b["batch_id"], 0, paid_at="")
self.assertEqual(before, self.ledger)
def test_batch_snapshot_last_paid_follows(self):
"""批次明细快照里的「最近收款月份」跟台账一起走,页面不会出现两个口径。"""
b = self._batch()
rid = b["documents"][0]["row_ids"][0]
b2, _ = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
det = [d for d in b2["details"] if d["row_id"] == rid][0]
self.assertEqual(det["last_paid"], "2026-10")
self.assertEqual(det["last_paid"], self.row(rid)["AB"])
def test_rollback_receipt_sync(self):
b = self._batch()
rid = b["documents"][0]["row_ids"][0]
self.row(rid)["AB"] = "2026-08"
b2, _ = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
self.assertEqual(self.row(rid)["AB"], "2026-10")
res = service.rollback_receipt_sync(b2)
self.assertGreaterEqual(res["reverted"], 1)
self.assertEqual(self.row(rid)["AB"], "2026-08")
def test_rollback_skips_manually_changed(self):
b = self._batch()
rid = b["documents"][0]["row_ids"][0]
b2, _ = service.set_receipt(b["batch_id"], 0, paid_at="2026-10-13")
self.row(rid)["AB"] = "2027-01" # 人工又改过
res = service.rollback_receipt_sync(b2)
self.assertEqual(res["reverted"], 0)
self.assertEqual(res["skipped"], len(b2.get("receipt_sync") or []))
self.assertEqual(self.row(rid)["AB"], "2027-01")
def test_elec_batch_does_not_touch_ledger(self):
"""电费批次不受影响(只房租车位费批次同步账期)。"""
b = {"batch_id": "E1", "kind": "elec", "documents": [
{"row_ids": ["r3"], "paid_at": ""}], "details": []}
self.batches["E1"] = b
_, sync = service.set_receipt("E1", 0, paid_at="2026-10-13")
self.assertEqual(sync["updated"], 0)
self.assertIsNone(self.row("r3")["AB"])
# ================================================================ 2. 批次数据导出
class TestExport(FakeBatchBackend):
def _batch(self):
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -421,6 +421,12 @@ td.mono { font-variant-numeric: tabular-nums; }
.banner.warn { background: #fffbeb; border-color: #fde68a; color: #92400e; }
.banner.ok { background: #f0fdf4; border-color: #bbf7d0; color: #166534; }
.banner.err { background: #fef2f2; border-color: #fecaca; color: #991b1b; }
.banner.ro { background: var(--brand-soft); border-color: var(--brand-line); color: var(--brand-d); }
/* ---- 只读访客:隐藏一切写操作入口(服务端另有硬拦截,这里只是不误导人) ---- */
body.role-viewer [data-ro] { display: none !important; }
#roChip { background: var(--brand-soft); border-color: var(--brand-line); color: var(--brand-d); }
#roChip .dot { background: var(--brand); }
.muted { color: var(--text-2); }
.mono-sm { font-size: 12px; font-variant-numeric: tabular-nums; }
......
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