133 lines
6.9 KiB
Python
133 lines
6.9 KiB
Python
|
"""A private family register, independent from login accounts and camera roles."""
|
||
|
|
import datetime as dt
|
||
|
|
import json
|
||
|
|
import secrets
|
||
|
|
|
||
|
|
|
||
|
|
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);
|
||
|
|
''')
|
||
|
|
self.a.DB.commit()
|
||
|
|
|
||
|
|
def authorize(self, user, edit=False):
|
||
|
|
if user['role'] != 'admin' 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:
|
||
|
|
return {'revision': self.a.setting('familyRevision', 0), 'people': self.a.objects('people'),
|
||
|
|
'links': self.a.objects('family_links'),
|
||
|
|
'canEdit': user['role'] == 'admin' or user.get('familyAccess') == 'edit',
|
||
|
|
'selfId': user.get('personId', '')}
|
||
|
|
|
||
|
|
def save(self, kind, data, user):
|
||
|
|
self.authorize(user, True)
|
||
|
|
a = self.a
|
||
|
|
with a.LOCK:
|
||
|
|
revision = a.setting('familyRevision', 0)
|
||
|
|
if type(data.get('revision')) is not int or data['revision'] != revision:
|
||
|
|
raise a.Problem('家谱已被更新，请重新载入后再编辑', 409)
|
||
|
|
table = 'people' if kind == 'person' else 'family_links'
|
||
|
|
old = a.get_object(table, data.get('id'))
|
||
|
|
if data.get('id') and not old:
|
||
|
|
raise a.Problem('家谱记录不存在', 404)
|
||
|
|
item = self.person(data, old) if kind == 'person' else self.link(data, old)
|
||
|
|
item.update(updatedAt=dt.datetime.now(dt.timezone.utc).isoformat(), updatedBy=user['username'])
|
||
|
|
# Revision and record are committed together, so concurrent forms cannot overwrite silently.
|
||
|
|
a.DB.execute('INSERT OR REPLACE INTO ' + table + ' VALUES (?,?)', (item['id'], json.dumps(item)))
|
||
|
|
a.DB.execute('INSERT OR REPLACE INTO settings VALUES (?,?)', ('familyRevision', json.dumps(revision + 1)))
|
||
|
|
a.DB.commit()
|
||
|
|
a.audit('更新家谱人物' if kind == 'person' else '更新家谱关系', user['username'])
|
||
|
|
return {'item': item, 'revision': revision + 1}
|
||
|
|
|
||
|
|
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')
|
||
|
|
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', 'unknown')
|
||
|
|
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('归档状态不正确')
|
||
|
|
return item
|
||
|
|
|
||
|
|
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))
|
||
|
|
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
|