"""A private family register, independent from login accounts and camera roles.""" import geography import datetime as dt import json import secrets from urllib.parse import urlsplit class Family: def __init__(self, app): self.a = app def initialize(self): with self.a.LOCK: self.a.DB.executescript(''' CREATE TABLE IF NOT EXISTS people (id TEXT PRIMARY KEY, body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS family_links (id TEXT PRIMARY KEY, body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS family_events (id TEXT PRIMARY KEY, body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS family_changes ( revision INTEGER PRIMARY KEY, kind TEXT NOT NULL, target_id TEXT NOT NULL, before_json TEXT, after_json TEXT NOT NULL, actor TEXT NOT NULL, at TEXT NOT NULL, restored_from INTEGER); CREATE INDEX IF NOT EXISTS family_changes_target ON family_changes(kind,target_id,revision); ''') self.a.DB.commit() def authorize(self, user, edit=False): self.a.HOUSEHOLDS.scope(user) if not self.a.HOUSEHOLDS.manager(user) and user.get('familyAccess', 'none') not in (('edit',) if edit else ('read', 'edit')): raise self.a.Problem('没有家谱' + ('编辑' if edit else '查看') + '权限', 403) def snapshot(self, user): self.authorize(user) with self.a.LOCK: can_edit = self.a.HOUSEHOLDS.manager(user) or user.get('familyAccess') == 'edit' people = self.a.HOUSEHOLDS.objects(user, 'people') return {'revision': self.a.setting(self.a.HOUSEHOLDS.revision_key(user), 0), 'people': people, 'links': self.a.HOUSEHOLDS.objects(user, 'family_links'), 'events': [e for e in self.a.HOUSEHOLDS.objects(user, 'family_events') if can_edit or (e.get('status') == 'published' and not e.get('archived'))], 'canEdit': can_edit, 'selfId': user.get('personId', '') if any(p['id'] == user.get('personId') for p in people) else '', 'familyId': self.a.HOUSEHOLDS.scope(user)} def save(self, kind, data, user, restored_from=None): self.authorize(user, True) a = self.a with a.LOCK: revision_key = a.HOUSEHOLDS.revision_key(user) revision = a.setting(revision_key, 0) if type(data.get('revision')) is not int or data['revision'] != revision: raise a.Problem('家谱已被更新，请重新载入后再编辑', 409) if kind not in ('person', 'link', 'event'): raise a.Problem('家谱记录类型不存在', 404) table = {'person': 'people', 'link': 'family_links', 'event': 'family_events'}[kind] family_id = a.HOUSEHOLDS.prepare(user, table, data, (('people', 'fromId'), ('people', 'toId')) if kind == 'link' else ()) if kind == 'event' and isinstance(data.get('personIds', []), list): for person in data.get('personIds', []): a.HOUSEHOLDS.owns(user, a.get_object('people', person)) old = a.get_object(table, data.get('id')) if data.get('id') and not old: raise a.Problem('家谱记录不存在', 404) item = {'person': self.person, 'link': self.link, 'event': self.event}[kind](data, old) item.update(familyId=family_id, updatedAt=dt.datetime.now(dt.timezone.utc).isoformat(), updatedBy=user['username']) if kind == 'event': item.update(createdAt=old.get('createdAt', item['updatedAt']) if old else item['updatedAt'], createdBy=old.get('createdBy', user['username']) if old else user['username']) # Withdrawal and restoration must retain the first publication time. if old and old.get('publishedAt'): item['publishedAt'] = old['publishedAt'] elif item['status'] == 'published': item['publishedAt'] = item['updatedAt'] # Revision and record are committed together, so concurrent forms cannot overwrite silently. with a.DB: change_revision = a.DB.execute('SELECT COALESCE(MAX(revision),0)+1 FROM family_changes').fetchone()[0] a.DB.execute('INSERT OR REPLACE INTO ' + table + ' VALUES (?,?)', (item['id'], json.dumps(item))) a.DB.execute('INSERT OR REPLACE INTO settings VALUES (?,?)', (revision_key, json.dumps(revision + 1))) a.DB.execute('INSERT INTO family_changes VALUES (?,?,?,?,?,?,?,?)', (change_revision, kind, item['id'], json.dumps(old) if old else None, json.dumps(item), user['username'], item['updatedAt'], restored_from)) a.audit({'person': '更新家谱人物', 'link': '更新家谱关系', 'event': '更新家族记事'}[kind], user['username']) return {'item': item, 'revision': revision + 1} def history(self, user, kind, target_id, before=None): # Previous drafts and withdrawn text are visible only to family editors. self.authorize(user, True) a = self.a table = {'person': 'people', 'link': 'family_links', 'event': 'family_events'}.get(kind) with a.LOCK: if not table or not a.get_object(table, target_id): raise a.Problem('家谱记录不存在', 404) a.HOUSEHOLDS.owns(user, a.get_object(table, target_id)) if before is not None: a.integer(before, 1, 2**53 - 1) rows = a.DB.execute('SELECT * FROM family_changes WHERE kind=? AND target_id=? AND revision 20 else None, revision=a.setting(a.HOUSEHOLDS.revision_key(user), 0)) def change(self, user, change_revision): self.authorize(user, True) self.a.integer(change_revision, 1, 2**53 - 1) with self.a.LOCK: row = self.a.DB.execute('SELECT * FROM family_changes WHERE revision=?', (change_revision,)).fetchone() if not row: raise self.a.Problem('修改记录不存在', 404) self.a.HOUSEHOLDS.owns(user, json.loads(row['after_json'])) return dict(revision=row['revision'], kind=row['kind'], targetId=row['target_id'], before=json.loads(row['before_json']) if row['before_json'] else None, after=json.loads(row['after_json']), at=row['at'], actor=row['actor']) def restore(self, data, user): self.authorize(user, True) with self.a.LOCK: change = self.change(user, data.get('changeRevision')) side = data.get('side') if side not in ('before', 'after') or change[side] is None: raise self.a.Problem('请选择有效的历史版本；创建前没有可恢复的内容') # Reuse all current relationship/date/publication validation. Restore is a new edit, # not an overwrite of history or a rollback of other people's records. return self.save(change['kind'], dict(change[side], revision=data.get('revision')), user, restored_from=change['revision']) def event(self, data, old): a = self.a item = {'id': old['id'] if old else secrets.token_hex(8)} for key, maximum, required in [('title', 160, True), ('eventDate', 10, True), ('sourceName', 160, False), ('sourceUrl', 2000, False)]: item[key] = a.clean_text(data.get(key, ''), maximum, required) try: if dt.date.fromisoformat(item['eventDate']).isoformat() != item['eventDate']: raise ValueError() except ValueError: raise a.Problem('记事日期请填写有效日期，格式为 YYYY-MM-DD') for key, maximum, label in [('summary', 400, '摘要'), ('body', 12000, '正文')]: value = data.get(key, '') if not isinstance(value, str) or len(value) > maximum or any(ord(c) < 32 and c not in '\r\n\t' for c in value): raise a.Problem(f'记事{label}格式不正确，最多 {maximum} 字') item[key] = value.strip() item['personIds'] = data.get('personIds', []) if (not isinstance(item['personIds'], list) or len(item['personIds']) > 200 or any(not isinstance(p, str) or not a.get_object('people', p) for p in item['personIds'])): raise a.Problem('请选择有效的关联人物，最多 200 位') item['personIds'] = list(dict.fromkeys(item['personIds'])) item['status'] = data.get('status', old.get('status', 'draft') if old else 'draft') item['archived'] = data.get('archived', old.get('archived', False) if old else False) if item['status'] not in ('draft', 'published') or not isinstance(item['archived'], bool): raise a.Problem('记事状态不正确') if item['sourceUrl']: try: parsed = urlsplit(item['sourceUrl']) if parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.username or parsed.password: raise ValueError() parsed.port except ValueError: raise a.Problem('报道链接请填写完整的 http 或 https 地址，不可包含账号密码') if item['status'] == 'published' and not (item['body'] or item['summary'] or item['sourceUrl']): raise a.Problem('发布前请填写记事正文、摘要或报道链接') return item def person(self, data, old): a = self.a item = {'id': old['id'] if old else secrets.token_hex(8)} for key, maximum in [('name', 80), ('alias', 120), ('birthDate', 10), ('deathDate', 10), ('birthplace', 160), ('biography', 2000), ('note', 500)]: value = data.get(key, '') if key in ('biography', 'note'): if not isinstance(value, str) or len(value) > maximum or any(ord(c) < 32 and c not in '\r\n\t' for c in value): raise a.Problem('生平或备注格式不正确') item[key] = value.strip() else: item[key] = a.clean_text(value, maximum, key == 'name') # Keep the legacy combined field verbatim. New structured locations are optional; # old clients updating another field must not erase them. for prefix in ('native', 'birth'): for field in ('Country', 'Province', 'City', 'District', 'Town', 'Community', 'Address'): key = prefix+field value = data.get(key, (old or {}).get(key, '')) if not isinstance(value, str): raise a.Problem('地域信息须为文字') item[key] = a.clean_text(value, 200 if field == 'Address' else 80) for prefix in ('native', 'birth'): geography.apply(item, data, old, a.clean_text, a.Problem, prefix) for key in ('birthDate', 'deathDate'): if item[key]: try: if dt.date.fromisoformat(item[key]).isoformat() != item[key]: raise ValueError() except ValueError: raise a.Problem('日期请使用公历 YYYY-MM-DD；不确定可留空，在备注记录') if item['birthDate'] and item['deathDate'] and item['birthDate'] > item['deathDate']: raise a.Problem('逝世日期不能早于出生日期') item['gender'] = data.get('gender', 'unknown') item['lifeStatus'] = data.get('lifeStatus', 'alive') if item['gender'] not in ('male', 'female', 'unknown') or item['lifeStatus'] not in ('alive', 'deceased', 'unknown'): raise a.Problem('人物属性不正确') if item['deathDate'] and item['lifeStatus'] != 'deceased': raise a.Problem('填写逝世日期时，请选择已故') item['archived'] = data.get('archived', False) if not isinstance(item['archived'], bool): raise a.Problem('归档状态不正确') if not old or item['birthDate'] != old.get('birthDate', ''): self.check_birth_order([e for e in a.objects('family_links') if item['id'] in (e['fromId'], e['toId'])], item) return item def check_birth_order(self, links, proposed_person=None): people = {p['id']: p for p in self.a.objects('people')} if proposed_person: people[proposed_person['id']] = proposed_person for edge in links: if not edge.get('active') or edge['kind'] != 'parent' or edge.get('lineage') != 'biological': continue parent, child = people.get(edge['fromId']), people.get(edge['toId']) if parent and child and parent.get('birthDate') and child.get('birthDate') and parent['birthDate'] >= child['birthDate']: raise self.a.Problem('亲生父母的出生日期必须早于子女，请核对日期或亲子属性') def link(self, data, old): a = self.a left, right = data.get('fromId'), data.get('toId') if not isinstance(left, str) or not isinstance(right, str) or not a.get_object('people', left) or not a.get_object('people', right): raise a.Problem('请先录入关系双方的人物') if left == right: raise a.Problem('不能把人物与自己建立亲属关系') kind = data.get('kind') if kind not in ('parent', 'spouse'): raise a.Problem('关系类型不正确') lineage = data.get('lineage', 'unspecified') if lineage not in ('biological', 'adoptive', 'step', 'unspecified'): raise a.Problem('亲子关系属性不正确') active = data.get('active', True) if not isinstance(active, bool): raise a.Problem('关系状态不正确') item = dict(id=old['id'] if old else secrets.token_hex(8), fromId=left, toId=right, kind=kind, lineage=lineage if kind == 'parent' else 'unspecified', active=active, note=a.clean_text(data.get('note', ''), 300)) self.check_birth_order([item]) links = [e for e in a.objects('family_links') if e.get('active') and e['id'] != item['id']] if active: for e in links: same = (e['fromId'], e['toId']) == (left, right) if kind == 'spouse': same = {e['fromId'], e['toId']} == {left, right} if e['kind'] == kind and same: raise a.Problem('这条关系已经存在', 409) graph = {} for e in links + [item]: if e['kind'] == 'parent': graph.setdefault(e['fromId'], []).append(e['toId']) # Iterative walk supports deep ancestry without Python recursion limits. def reaches(start, target): pending, seen = [start], set() while pending: p = pending.pop() if p in seen: continue seen.add(p) for child in graph.get(p, []): if child == target: return True pending.append(child) return False if kind == 'parent' and reaches(right, left): raise a.Problem('这条关系会形成祖先与后代循环，请检查方向') for e in links + [item]: if e['kind'] == 'spouse' and (reaches(e['fromId'], e['toId']) or reaches(e['toId'], e['fromId'])): raise a.Problem('配偶不能同时是直系祖先或后代，请检查已有关系') return item