Spaces:

plokmii
/

pradsa

Sleeping

pradsa / sync.py

Upload sync.py with huggingface_hub

9b2a2fc verified 7 months ago

11.5 kB

# -*- coding: utf-8 -*-
"""
雙向同步：
  - pull:  雲端 → Local（雲端為主，覆蓋 Local）
  - push:  Local → 雲端（Append-only merge，不刪雲端資料）
用法: python sync.py          # pull（預設）
      python sync.py push     # push
"""

import configparser
import sqlite3
import shutil
import os
import tempfile
from pathlib import Path
from datetime import datetime

TOKEN = os.environ.get('HF_TOKEN', '')
USERNAME = os.environ.get('HF_USERNAME', '')
if not TOKEN:
    cfg = configparser.ConfigParser()
    cfg.read('Myconfig.ini')
    try:
        TOKEN = cfg['Hugging Face']['Token'].strip()
        USERNAME = cfg['Hugging Face']['Username'].strip()
    except KeyError:
        pass
REPO_ID = f'{USERNAME}/pradsa'

LOCAL_DB = Path('pra_data.db')
BACKUP_DIR = LOCAL_DB.parent / 'backups'


def get_db_info(db_path):
    if not Path(db_path).exists():
        return {'patients': 0, 'reports': 0, 'users': 0, 'latest': ''}
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    try:
        patients = conn.execute('SELECT COUNT(*) as c FROM patients').fetchone()['c']
        reports = conn.execute('SELECT COUNT(*) as c FROM reports WHERE COALESCE(is_deleted,0)=0').fetchone()['c']
        users = conn.execute('SELECT COUNT(*) as c FROM users').fetchone()['c']
        latest = conn.execute(
            "SELECT MAX(COALESCE(updated_at, created_at)) as t FROM reports"
        ).fetchone()['t'] or ''
    except Exception:
        patients, reports, users, latest = 0, 0, 0, ''
    conn.close()
    return {'patients': patients, 'reports': reports, 'users': users, 'latest': latest}


def download_cloud_db():
    from huggingface_hub import hf_hub_download
    try:
        path = hf_hub_download(
            repo_id=REPO_ID, filename='pra_data.db',
            repo_type='space', token=TOKEN,
            local_dir=tempfile.gettempdir(),
            force_download=True,
        )
        return path
    except Exception as e:
        print(f'  雲端 DB 下載失敗: {e}')
        return None


def upload_db(db_path):
    from huggingface_hub import HfApi
    api = HfApi(token=TOKEN)
    api.upload_file(
        path_or_fileobj=str(db_path),
        path_in_repo='pra_data.db',
        repo_id=REPO_ID,
        repo_type='space',
    )


def _backup_local(tag='sync'):
    """備份 Local DB（當天最多 5 個，總共最多 15 個）"""
    if not LOCAL_DB.exists():
        return None
    BACKUP_DIR.mkdir(exist_ok=True)
    today = datetime.now().strftime('%Y%m%d')
    ts = datetime.now().strftime('%Y%m%d_%H%M%S')
    bk = BACKUP_DIR / f'pra_data_before_{tag}_{ts}.db'
    shutil.copy2(str(LOCAL_DB), str(bk))
    print(f'已備份 Local → {bk.name}')

    today_bks = sorted(BACKUP_DIR.glob(f'pra_data_before_{tag}_{today}_*.db'))
    for old in today_bks[:-5]:
        old.unlink()
    all_bks = sorted(BACKUP_DIR.glob(f'pra_data_before_{tag}_*.db'))
    for old in all_bks[:-15]:
        old.unlink()
    return bk


def merge_local_into_cloud(local_path, cloud_path, merged_path):
    """
    Append-only merge：以雲端為底，把 Local 的新增/更新資料合併進去。
    絕不刪除雲端已有的 patients / reports / users。
    """
    shutil.copy2(cloud_path, merged_path)

    merged = sqlite3.connect(str(merged_path))
    merged.row_factory = sqlite3.Row
    local = sqlite3.connect(str(local_path))
    local.row_factory = sqlite3.Row

    # --- patients: 以 chart_no 為 key，UPSERT ---
    local_patients = local.execute('SELECT * FROM patients').fetchall()
    for p in local_patients:
        existing = merged.execute('SELECT id FROM patients WHERE chart_no=?', (p['chart_no'],)).fetchone()
        if existing:
            # 更新 patient_name, donor_hla（取較新的）
            merged.execute('UPDATE patients SET patient_name=?, donor_hla=? WHERE chart_no=?',
                           (p['patient_name'], p['donor_hla'] or '', p['chart_no']))
        else:
            merged.execute(
                'INSERT INTO patients (patient_name, chart_no, created_at, donor_hla) VALUES (?,?,?,?)',
                (p['patient_name'], p['chart_no'], p['created_at'], p['donor_hla'] or ''))

    # 建立 chart_no → merged patient_id 對照
    merged_patients = {r['chart_no']: r['id'] for r in merged.execute('SELECT id, chart_no FROM patients').fetchall()}
    # 建立 local patient_id → chart_no 對照
    local_pt_map = {r['id']: r['chart_no'] for r in local.execute('SELECT id, chart_no FROM patients').fetchall()}

    # --- reports: 以 (patient chart_no + report_date + pra_class) 為 key ---
    local_reports = local.execute('SELECT * FROM reports').fetchall()
    for r in local_reports:
        chart_no = local_pt_map.get(r['patient_id'])
        if not chart_no:
            continue
        merged_pid = merged_patients.get(chart_no)
        if not merged_pid:
            continue

        existing = merged.execute(
            'SELECT id, COALESCE(is_deleted,0) as is_deleted, updated_at FROM reports WHERE patient_id=? AND report_date=? AND pra_class=?',
            (merged_pid, r['report_date'], r['pra_class'])
        ).fetchone()

        if existing:
            # 雲端有此 report → 更新內容（但不把雲端 active 的改為 deleted）
            cloud_deleted = existing['is_deleted']
            local_deleted = r['is_deleted'] if r['is_deleted'] is not None else 0
            # 如果 local 標記刪除但雲端沒有 → 保留雲端的 active 狀態
            new_deleted = 0 if (cloud_deleted == 0 and local_deleted == 1) else local_deleted
            merged.execute('''UPDATE reports SET pra_percent=?, overall=?, specificity=?, comment=?,
                              status=?, submitted_by=?, upload_file=?, updated_at=?,
                              is_deleted=?, deleted_at=? WHERE id=?''',
                           (r['pra_percent'], r['overall'], r['specificity'], r['comment'],
                            r['status'] or 'draft', r['submitted_by'] or '',
                            r['upload_file'] or '', r['updated_at'],
                            new_deleted, r['deleted_at'] if new_deleted else None,
                            existing['id']))
            merged_rid = existing['id']
        else:
            # 雲端沒有 → 新增
            cur = merged.execute(
                '''INSERT INTO reports (patient_id, report_date, pra_class, pra_percent, overall,
                   specificity, comment, status, submitted_by, upload_file, created_at, updated_at,
                   is_deleted, deleted_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)''',
                (merged_pid, r['report_date'], r['pra_class'], r['pra_percent'], r['overall'],
                 r['specificity'], r['comment'], r['status'] or 'draft',
                 r['submitted_by'] or '', r['upload_file'] or '',
                 r['created_at'], r['updated_at'],
                 r['is_deleted'] if r['is_deleted'] is not None else 0, r['deleted_at']))
            merged_rid = cur.lastrowid

        # antibody_strength: 用 local 的資料覆蓋（屬於同一 report）
        local_rid = r['id']
        local_abs = local.execute('SELECT * FROM antibody_strength WHERE report_id=?', (local_rid,)).fetchall()
        if local_abs:
            merged.execute('DELETE FROM antibody_strength WHERE report_id=?', (merged_rid,))
            for a in local_abs:
                merged.execute(
                    '''INSERT INTO antibody_strength (report_id, antigen, allele, max_mfi, mean_mfi, no_of_beads, bead_ids)
                       VALUES (?,?,?,?,?,?,?)''',
                    (merged_rid, a['antigen'], a['allele'], a['max_mfi'], a['mean_mfi'],
                     a['no_of_beads'], a['bead_ids']))

    # --- users: 以 username 為 key，只新增不刪 ---
    local_users = local.execute('SELECT * FROM users').fetchall()
    for u in local_users:
        existing = merged.execute('SELECT id FROM users WHERE username=?', (u['username'],)).fetchone()
        if existing:
            # 更新 display_name, role, password（取 local 版本）
            merged.execute('UPDATE users SET display_name=?, password_hash=?, password_plain=?, role=?, edit_time=? WHERE id=?',
                           (u['display_name'] or '', u['password_hash'], u['password_plain'] or '',
                            u['role'] or 'user', u['edit_time'], existing['id']))
        else:
            merged.execute(
                'INSERT INTO users (username, display_name, password_hash, password_plain, role, created_at, edit_time) VALUES (?,?,?,?,?,?,?)',
                (u['username'], u['display_name'] or '', u['password_hash'],
                 u['password_plain'] or '', u['role'] or 'user', u['created_at'], u['edit_time']))

    merged.commit()
    local.close()
    merged.close()


def push():
    """Local → 雲端 (Append-only merge)"""
    print('=== Local → 雲端 Append-only Push ===\n')

    local_info = get_db_info(LOCAL_DB)
    print(f'Local:  {local_info["patients"]} 病患, {local_info["reports"]} 報告, {local_info["users"]} 使用者')

    print('下載雲端 DB...')
    cloud_path = download_cloud_db()
    if not cloud_path:
        print('無法取得雲端 DB，改為直接上傳 Local DB')
        upload_db(LOCAL_DB)
        print('已直接上傳')
        return

    cloud_info = get_db_info(cloud_path)
    print(f'Cloud:  {cloud_info["patients"]} 病患, {cloud_info["reports"]} 報告, {cloud_info["users"]} 使用者')
    print()

    # Merge
    merged_path = Path(tempfile.gettempdir()) / 'pra_data_merged.db'
    merge_local_into_cloud(str(LOCAL_DB), cloud_path, str(merged_path))

    merged_info = get_db_info(merged_path)
    print(f'Merged: {merged_info["patients"]} 病患, {merged_info["reports"]} 報告, {merged_info["users"]} 使用者')

    # Upload merged
    upload_db(merged_path)
    print('\n已推送 merged DB 到雲端')

    # 也更新 Local 為 merged 版本
    _backup_local('push')
    shutil.copy2(str(merged_path), str(LOCAL_DB))
    print('Local DB 已同步為 merged 版本')


def sync():
    """雲端 → Local（pull，雲端為主）"""
    print('=== 雲端 → Local 單向同步 ===\n')

    local_info = get_db_info(LOCAL_DB)
    print(f'Local:  {local_info["patients"]} 病患, {local_info["reports"]} 報告, {local_info["users"]} 使用者, latest: {local_info["latest"] or "(none)"}')

    print('下載雲端 DB...')
    cloud_path = download_cloud_db()
    if not cloud_path:
        print('無法取得雲端 DB')
        return

    cloud_info = get_db_info(cloud_path)
    print(f'Cloud:  {cloud_info["patients"]} 病患, {cloud_info["reports"]} 報告, {cloud_info["users"]} 使用者, latest: {cloud_info["latest"] or "(none)"}')
    print()

    if (local_info['latest'] == cloud_info['latest']
            and local_info['reports'] == cloud_info['reports']
            and local_info['patients'] == cloud_info['patients']
            and local_info['users'] == cloud_info['users']):
        print('兩邊一致，不需要同步')
        return

    _backup_local('sync')

    # 直接用雲端 DB 覆蓋 Local
    shutil.copy2(cloud_path, str(LOCAL_DB))

    new_info = get_db_info(LOCAL_DB)
    print(f'\n同步完成: {new_info["patients"]} 病患, {new_info["reports"]} 報告, {new_info["users"]} 使用者')


if __name__ == '__main__':
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == 'push':
        push()
    else:
        sync()

	# -- coding: utf-8 --
	"""
	雙向同步：
	- pull: 雲端 → Local（雲端為主，覆蓋 Local）
	- push: Local → 雲端（Append-only merge，不刪雲端資料）
	用法: python sync.py # pull（預設）
	python sync.py push # push
	"""

	import configparser
	import sqlite3
	import shutil
	import os
	import tempfile
	from pathlib import Path
	from datetime import datetime

	TOKEN = os.environ.get('HF_TOKEN', '')
	USERNAME = os.environ.get('HF_USERNAME', '')
	if not TOKEN:
	cfg = configparser.ConfigParser()
	cfg.read('Myconfig.ini')
	try:
	TOKEN = cfg['Hugging Face']['Token'].strip()
	USERNAME = cfg['Hugging Face']['Username'].strip()
	except KeyError:
	pass
	REPO_ID = f'{USERNAME}/pradsa'

	LOCAL_DB = Path('pra_data.db')
	BACKUP_DIR = LOCAL_DB.parent / 'backups'


	def get_db_info(db_path):
	if not Path(db_path).exists():
	return {'patients': 0, 'reports': 0, 'users': 0, 'latest': ''}
	conn = sqlite3.connect(str(db_path))
	conn.row_factory = sqlite3.Row
	try:
	patients = conn.execute('SELECT COUNT(*) as c FROM patients').fetchone()['c']
	reports = conn.execute('SELECT COUNT(*) as c FROM reports WHERE COALESCE(is_deleted,0)=0').fetchone()['c']
	users = conn.execute('SELECT COUNT(*) as c FROM users').fetchone()['c']
	latest = conn.execute(
	"SELECT MAX(COALESCE(updated_at, created_at)) as t FROM reports"
	).fetchone()['t'] or ''
	except Exception:
	patients, reports, users, latest = 0, 0, 0, ''
	conn.close()
	return {'patients': patients, 'reports': reports, 'users': users, 'latest': latest}


	def download_cloud_db():
	from huggingface_hub import hf_hub_download
	try:
	path = hf_hub_download(
	repo_id=REPO_ID, filename='pra_data.db',
	repo_type='space', token=TOKEN,
	local_dir=tempfile.gettempdir(),
	force_download=True,
	)
	return path
	except Exception as e:
	print(f' 雲端 DB 下載失敗: {e}')
	return None


	def upload_db(db_path):
	from huggingface_hub import HfApi
	api = HfApi(token=TOKEN)
	api.upload_file(
	path_or_fileobj=str(db_path),
	path_in_repo='pra_data.db',
	repo_id=REPO_ID,
	repo_type='space',
	)


	def _backup_local(tag='sync'):
	"""備份 Local DB（當天最多 5 個，總共最多 15 個）"""
	if not LOCAL_DB.exists():
	return None
	BACKUP_DIR.mkdir(exist_ok=True)
	today = datetime.now().strftime('%Y%m%d')
	ts = datetime.now().strftime('%Y%m%d_%H%M%S')
	bk = BACKUP_DIR / f'pra_data_before_{tag}_{ts}.db'
	shutil.copy2(str(LOCAL_DB), str(bk))
	print(f'已備份 Local → {bk.name}')

	today_bks = sorted(BACKUP_DIR.glob(f'pra_data_before_{tag}_{today}_*.db'))
	for old in today_bks[:-5]:
	old.unlink()
	all_bks = sorted(BACKUP_DIR.glob(f'pra_data_before_{tag}_*.db'))
	for old in all_bks[:-15]:
	old.unlink()
	return bk


	def merge_local_into_cloud(local_path, cloud_path, merged_path):
	"""
	Append-only merge：以雲端為底，把 Local 的新增/更新資料合併進去。
	絕不刪除雲端已有的 patients / reports / users。
	"""
	shutil.copy2(cloud_path, merged_path)

	merged = sqlite3.connect(str(merged_path))
	merged.row_factory = sqlite3.Row
	local = sqlite3.connect(str(local_path))
	local.row_factory = sqlite3.Row

	# --- patients: 以 chart_no 為 key，UPSERT ---
	local_patients = local.execute('SELECT * FROM patients').fetchall()
	for p in local_patients:
	existing = merged.execute('SELECT id FROM patients WHERE chart_no=?', (p['chart_no'],)).fetchone()
	if existing:
	# 更新 patient_name, donor_hla（取較新的）
	merged.execute('UPDATE patients SET patient_name=?, donor_hla=? WHERE chart_no=?',
	(p['patient_name'], p['donor_hla'] or '', p['chart_no']))
	else:
	merged.execute(
	'INSERT INTO patients (patient_name, chart_no, created_at, donor_hla) VALUES (?,?,?,?)',
	(p['patient_name'], p['chart_no'], p['created_at'], p['donor_hla'] or ''))

	# 建立 chart_no → merged patient_id 對照
	merged_patients = {r['chart_no']: r['id'] for r in merged.execute('SELECT id, chart_no FROM patients').fetchall()}
	# 建立 local patient_id → chart_no 對照
	local_pt_map = {r['id']: r['chart_no'] for r in local.execute('SELECT id, chart_no FROM patients').fetchall()}

	# --- reports: 以 (patient chart_no + report_date + pra_class) 為 key ---
	local_reports = local.execute('SELECT * FROM reports').fetchall()
	for r in local_reports:
	chart_no = local_pt_map.get(r['patient_id'])
	if not chart_no:
	continue
	merged_pid = merged_patients.get(chart_no)
	if not merged_pid:
	continue

	existing = merged.execute(
	'SELECT id, COALESCE(is_deleted,0) as is_deleted, updated_at FROM reports WHERE patient_id=? AND report_date=? AND pra_class=?',
	(merged_pid, r['report_date'], r['pra_class'])
	).fetchone()

	if existing:
	# 雲端有此 report → 更新內容（但不把雲端 active 的改為 deleted）
	cloud_deleted = existing['is_deleted']
	local_deleted = r['is_deleted'] if r['is_deleted'] is not None else 0
	# 如果 local 標記刪除但雲端沒有 → 保留雲端的 active 狀態
	new_deleted = 0 if (cloud_deleted == 0 and local_deleted == 1) else local_deleted
	merged.execute('''UPDATE reports SET pra_percent=?, overall=?, specificity=?, comment=?,
	status=?, submitted_by=?, upload_file=?, updated_at=?,
	is_deleted=?, deleted_at=? WHERE id=?''',
	(r['pra_percent'], r['overall'], r['specificity'], r['comment'],
	r['status'] or 'draft', r['submitted_by'] or '',
	r['upload_file'] or '', r['updated_at'],
	new_deleted, r['deleted_at'] if new_deleted else None,
	existing['id']))
	merged_rid = existing['id']
	else:
	# 雲端沒有 → 新增
	cur = merged.execute(
	'''INSERT INTO reports (patient_id, report_date, pra_class, pra_percent, overall,
	specificity, comment, status, submitted_by, upload_file, created_at, updated_at,
	is_deleted, deleted_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)''',
	(merged_pid, r['report_date'], r['pra_class'], r['pra_percent'], r['overall'],
	r['specificity'], r['comment'], r['status'] or 'draft',
	r['submitted_by'] or '', r['upload_file'] or '',
	r['created_at'], r['updated_at'],
	r['is_deleted'] if r['is_deleted'] is not None else 0, r['deleted_at']))
	merged_rid = cur.lastrowid

	# antibody_strength: 用 local 的資料覆蓋（屬於同一 report）
	local_rid = r['id']
	local_abs = local.execute('SELECT * FROM antibody_strength WHERE report_id=?', (local_rid,)).fetchall()
	if local_abs:
	merged.execute('DELETE FROM antibody_strength WHERE report_id=?', (merged_rid,))
	for a in local_abs:
	merged.execute(
	'''INSERT INTO antibody_strength (report_id, antigen, allele, max_mfi, mean_mfi, no_of_beads, bead_ids)
	VALUES (?,?,?,?,?,?,?)''',
	(merged_rid, a['antigen'], a['allele'], a['max_mfi'], a['mean_mfi'],
	a['no_of_beads'], a['bead_ids']))

	# --- users: 以 username 為 key，只新增不刪 ---
	local_users = local.execute('SELECT * FROM users').fetchall()
	for u in local_users:
	existing = merged.execute('SELECT id FROM users WHERE username=?', (u['username'],)).fetchone()
	if existing:
	# 更新 display_name, role, password（取 local 版本）
	merged.execute('UPDATE users SET display_name=?, password_hash=?, password_plain=?, role=?, edit_time=? WHERE id=?',
	(u['display_name'] or '', u['password_hash'], u['password_plain'] or '',
	u['role'] or 'user', u['edit_time'], existing['id']))
	else:
	merged.execute(
	'INSERT INTO users (username, display_name, password_hash, password_plain, role, created_at, edit_time) VALUES (?,?,?,?,?,?,?)',
	(u['username'], u['display_name'] or '', u['password_hash'],
	u['password_plain'] or '', u['role'] or 'user', u['created_at'], u['edit_time']))

	merged.commit()
	local.close()
	merged.close()


	def push():
	"""Local → 雲端 (Append-only merge)"""
	print('=== Local → 雲端 Append-only Push ===\n')

	local_info = get_db_info(LOCAL_DB)
	print(f'Local: {local_info["patients"]} 病患, {local_info["reports"]} 報告, {local_info["users"]} 使用者')

	print('下載雲端 DB...')
	cloud_path = download_cloud_db()
	if not cloud_path:
	print('無法取得雲端 DB，改為直接上傳 Local DB')
	upload_db(LOCAL_DB)
	print('已直接上傳')
	return

	cloud_info = get_db_info(cloud_path)
	print(f'Cloud: {cloud_info["patients"]} 病患, {cloud_info["reports"]} 報告, {cloud_info["users"]} 使用者')
	print()

	# Merge
	merged_path = Path(tempfile.gettempdir()) / 'pra_data_merged.db'
	merge_local_into_cloud(str(LOCAL_DB), cloud_path, str(merged_path))

	merged_info = get_db_info(merged_path)
	print(f'Merged: {merged_info["patients"]} 病患, {merged_info["reports"]} 報告, {merged_info["users"]} 使用者')

	# Upload merged
	upload_db(merged_path)
	print('\n已推送 merged DB 到雲端')

	# 也更新 Local 為 merged 版本
	_backup_local('push')
	shutil.copy2(str(merged_path), str(LOCAL_DB))
	print('Local DB 已同步為 merged 版本')


	def sync():
	"""雲端 → Local（pull，雲端為主）"""
	print('=== 雲端 → Local 單向同步 ===\n')

	local_info = get_db_info(LOCAL_DB)
	print(f'Local: {local_info["patients"]} 病患, {local_info["reports"]} 報告, {local_info["users"]} 使用者, latest: {local_info["latest"] or "(none)"}')

	print('下載雲端 DB...')
	cloud_path = download_cloud_db()
	if not cloud_path:
	print('無法取得雲端 DB')
	return

	cloud_info = get_db_info(cloud_path)
	print(f'Cloud: {cloud_info["patients"]} 病患, {cloud_info["reports"]} 報告, {cloud_info["users"]} 使用者, latest: {cloud_info["latest"] or "(none)"}')
	print()

	if (local_info['latest'] == cloud_info['latest']
	and local_info['reports'] == cloud_info['reports']
	and local_info['patients'] == cloud_info['patients']
	and local_info['users'] == cloud_info['users']):
	print('兩邊一致，不需要同步')
	return

	_backup_local('sync')

	# 直接用雲端 DB 覆蓋 Local
	shutil.copy2(cloud_path, str(LOCAL_DB))

	new_info = get_db_info(LOCAL_DB)
	print(f'\n同步完成: {new_info["patients"]} 病患, {new_info["reports"]} 報告, {new_info["users"]} 使用者')


	if __name__ == '__main__':
	import sys
	if len(sys.argv) > 1 and sys.argv[1] == 'push':
	push()
	else:
	sync()