digital-psychology/scripts/import_java_consult.py

#!/usr/bin/env python3
"""Pull public consult content + images from live Java into local PostgreSQL."""

from __future__ import annotations

import json
import os
import re
import ssl
import sys
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import quote, unquote, urlparse

JAVA = os.environ.get("JAVA_CONSULT_BASE", "https://miniapp.yuxingu.com.cn")
PG_DSN = os.environ.get(
    "DATABASE_URL",
    "postgres://yuxingu:yuxingu@127.0.0.1:5432/yuxingu?sslmode=disable",
)
ROOT = Path(__file__).resolve().parents[1] / "apps" / "api" / "data" / "yxg-mp"
CTX = ssl.create_default_context()

IMG_RE = re.compile(r"https://miniapp\.yuxingu\.com\.cn/yxg-mp/[^\\\"'\s)]+", re.I)


def api(path: str, method: str = "GET") -> dict:
    url = JAVA.rstrip("/") + path
    req = urllib.request.Request(url, method=method, headers={"Content-Type": "application/json"})
    if method == "POST":
        req.data = b"{}"
    with urllib.request.urlopen(req, timeout=30, context=CTX) as resp:
        return json.load(resp)


def get_data(path: str, method: str = "GET"):
    body = api(path, method)
    if str(body.get("code")) not in ("0", "0.0"):
        raise RuntimeError(f"{path} code={body.get('code')} msg={body.get('msg')}")
    return body.get("data")


def collect_urls(*blobs: object) -> set[str]:
    found: set[str] = set()
    for blob in blobs:
        found.update(IMG_RE.findall(json.dumps(blob, ensure_ascii=False)))
    return found


def download(url: str) -> None:
    url = url.rstrip("\\").rstrip()
    parsed = urlparse(url)
    rel = unquote(parsed.path)
    if not rel.startswith("/yxg-mp/"):
        return
    dest = ROOT / rel[len("/yxg-mp/") :]
    dest.parent.mkdir(parents=True, exist_ok=True)
    if dest.exists() and dest.stat().st_size > 0:
        return
    fetch = f"{parsed.scheme}://{parsed.netloc}{quote(rel, safe='/@')}"
    req = urllib.request.Request(fetch, headers={"User-Agent": "yxg-import/1"})
    with urllib.request.urlopen(req, timeout=60, context=CTX) as resp, dest.open("wb") as out:
        out.write(resp.read())
    print("saved", dest.relative_to(ROOT.parent), dest.stat().st_size)


def main() -> int:
    try:
        import psycopg
    except ImportError:
        os.system(f"{sys.executable} -m pip install 'psycopg[binary]' -q")
        import psycopg

    banners = get_data("/app-api/psychic/banner/all", "POST") or []
    news_items = []
    for typ in (1, 2):
        page = get_data(f"/app-api/psychic/news/page?type={typ}&pageNo=1&pageSize=50") or {}
        for row in page.get("list") or []:
            detail = get_data(f"/app-api/psychic/news/get?id={row['id']}") or row
            news_items.append(detail)
    doctors = []
    page = get_data("/app-api/psychic/doctor-info/page?pageNo=1&pageSize=50") or {}
    top_ids = {
        str(x.get("id"))
        for x in (get_data("/app-api/psychic/doctor-info/page?pageNo=1&pageSize=50&isTop=1") or {}).get("list") or []
    }
    for row in page.get("list") or []:
        detail = get_data(f"/app-api/psychic/doctor-info/get?id={row['id']}") or row
        detail["_is_top"] = 1 if str(detail.get("id")) in top_ids else 0
        doctors.append(detail)
    tests = get_data("/app-api/psychic/study/list", "POST") or []
    scopes = get_data("/app-api/psychic/doctor-info/business-scope-list") or []

    urls = collect_urls(banners, news_items, doctors, tests)
    print(f"content banners={len(banners)} news={len(news_items)} doctors={len(doctors)} tests={len(tests)} images={len(urls)}")
    ROOT.mkdir(parents=True, exist_ok=True)
    for url in sorted(urls):
        try:
            download(url)
        except urllib.error.HTTPError as e:
            print("skip", url, e.code)
        except Exception as e:
            print("skip", url, e)

    with psycopg.connect(PG_DSN) as conn:
        conn.execute(
            """
            TRUNCATE consult_slots, consult_schedule_days, consult_focus,
              consult_orders, consult_user_choices, consult_options,
              consult_questions, consult_test_results, consult_tests,
              consult_doctors, consult_news, consult_banners, consult_business_scopes
            RESTART IDENTITY CASCADE
            """
        )
        for s in scopes:
            conn.execute(
                "INSERT INTO consult_business_scopes(code, name) VALUES (%s, %s) ON CONFLICT (code) DO UPDATE SET name=EXCLUDED.name",
                (s.get("code") or "", s.get("name") or ""),
            )
        for b in banners:
            conn.execute(
                """INSERT INTO consult_banners(id, banner_name, banner_image, click_url, describe, order_index, jump_type)
                   VALUES (%s,%s,%s,%s,%s,%s,%s)""",
                (
                    int(b["id"]),
                    b.get("bannerName") or "",
                    b.get("bannerImage") or "",
                    b.get("clickUrl") or "",
                    b.get("describe") or "",
                    int(b.get("orderIndex") or 0),
                    int(b.get("jumpType") or 1),
                ),
            )
        for n in news_items:
            conn.execute(
                """INSERT INTO consult_news(id, type, title, show_image, content, show_main)
                   VALUES (%s,%s,%s,%s,%s,%s)""",
                (
                    int(n["id"]),
                    int(n.get("type") or 0),
                    n.get("title") or "",
                    n.get("showImage") or "",
                    n.get("content") or "",
                    int(n.get("showMain") or 0),
                ),
            )
        for d in doctors:
            conn.execute(
                """INSERT INTO consult_doctors(
                     id, name, avatar, business_scope, cover_url, consultation_method, education,
                     introduction, resume, notice, tags, work_experience, work_start_time, price,
                     status, address, address_detail, is_top)
                   VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,%s,%s,%s)""",
                (
                    int(d["id"]),
                    d.get("name") or "",
                    d.get("avatar") or "",
                    d.get("businessScope") or "",
                    d.get("coverUrl") or d.get("avatar") or "",
                    d.get("consultationMethod") or "online",
                    d.get("education") or "",
                    d.get("introduction") or "",
                    d.get("resume") or "",
                    d.get("notice") or "",
                    d.get("tags") or "",
                    d.get("workExperience") or "",
                    d.get("workStartTime") or None,
                    int(d.get("price") or 0),
                    d.get("address") or "",
                    d.get("addressDetail") or "",
                    int(d.get("_is_top") or 0),
                ),
            )
        for t in tests:
            conn.execute(
                """INSERT INTO consult_tests(id, test_name, sub_title, test_pic, total_num, show_main, status)
                   VALUES (%s,%s,%s,%s,%s,%s,1)""",
                (
                    int(t["id"]),
                    t.get("testName") or "",
                    t.get("subTitle") or "",
                    t.get("testPic") or "",
                    int(t.get("totalNum") or 0),
                    int(t.get("showMain") or 0),
                ),
            )
        conn.execute(
            """
            INSERT INTO consult_schedule_days(doctor_id, schedule_date)
            SELECT d.id, (CURRENT_DATE + g.n)
            FROM consult_doctors d
            CROSS JOIN generate_series(1, 14) AS g(n)
            ON CONFLICT (doctor_id, schedule_date) DO NOTHING
            """
        )
        conn.execute(
            """
            INSERT INTO consult_slots(schedule_id, doctor_id, start_time, end_time, consultation_method, status)
            SELECT s.id, s.doctor_id, t.st, t.et, 'online', 0
            FROM consult_schedule_days s
            CROSS JOIN (VALUES (TIME '09:00', TIME '10:00'), (TIME '14:00', TIME '15:00')) AS t(st, et)
            WHERE NOT EXISTS (
              SELECT 1 FROM consult_slots x WHERE x.schedule_id=s.id AND x.start_time=t.st
            )
            """
        )
        conn.execute(
            """
            INSERT INTO consult_tests(test_name, sub_title, test_pic, test_introduction, test_notice, total_num, show_main, status)
            SELECT '情绪小测', '用两分钟看看最近的状态',
                   'https://miniapp.yuxingu.com.cn/yxg-mp/2025/06/27/生活满意度指数A_0.jpg',
                   '本测评仅供自我觉察，不是诊断。', '请按第一直觉作答。', 1280, 1, 1
            WHERE NOT EXISTS (SELECT 1 FROM consult_tests WHERE test_name='情绪小测')
            """
        )
        conn.execute(
            """
            INSERT INTO consult_questions(test_id, question_type, question_text, required, order_index)
            SELECT t.id, 1, q.txt, 1, q.ord
            FROM consult_tests t
            CROSS JOIN (VALUES
              ('最近两周，我感到紧张或坐立不安。', 1),
              ('最近两周，我仍然能享受日常小事。', 2)
            ) AS q(txt, ord)
            WHERE t.test_name='情绪小测'
              AND NOT EXISTS (SELECT 1 FROM consult_questions x WHERE x.test_id=t.id)
            """
        )
        conn.execute(
            """
            INSERT INTO consult_options(question_id, option_text, option_score, order_index)
            SELECT q.id, o.txt, o.sc, o.ord
            FROM consult_questions q
            JOIN consult_tests t ON t.id=q.test_id AND t.test_name='情绪小测'
            CROSS JOIN (VALUES ('很少', 0, 1), ('有时', 1, 2), ('经常', 2, 3)) AS o(txt, sc, ord)
            WHERE NOT EXISTS (SELECT 1 FROM consult_options x WHERE x.question_id=q.id)
            """
        )
        conn.execute(
            """
            INSERT INTO consult_test_results(test_id, min_score, max_score, result_desc, result_analysis, treat_plan)
            SELECT t.id, r.a, r.b, r.d, r.an, r.p
            FROM consult_tests t
            CROSS JOIN (VALUES
              (0, 2, '状态平稳', '你最近的情绪波动在可调节范围内。', '保持作息，需要时可以和咨询师聊聊。'),
              (3, 8, '需要被看见', '最近的紧绷感偏高，适合做一次梳理。', '建议预约咨询，或先做呼吸放松。')
            ) AS r(a,b,d,an,p)
            WHERE t.test_name='情绪小测'
              AND NOT EXISTS (SELECT 1 FROM consult_test_results x WHERE x.test_id=t.id)
            """
        )
        conn.execute("SELECT setval(pg_get_serial_sequence('consult_banners', 'id'), COALESCE((SELECT MAX(id) FROM consult_banners), 1))")
        conn.execute("SELECT setval(pg_get_serial_sequence('consult_news', 'id'), COALESCE((SELECT MAX(id) FROM consult_news), 1))")
        conn.execute("SELECT setval(pg_get_serial_sequence('consult_doctors', 'id'), COALESCE((SELECT MAX(id) FROM consult_doctors), 1))")
        conn.execute("SELECT setval(pg_get_serial_sequence('consult_tests', 'id'), COALESCE((SELECT MAX(id) FROM consult_tests), 1))")
        conn.commit()
    print("import ok")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())