# -*- coding: utf-8 -*- """SmartPaper R3-Free UI: Streamlit is an observer, never the paper executor.""" import hashlib import json import logging import os import re import secrets import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path import streamlit as st import streamlit.components.v1 as components from config import APP_CONFIG from managed_auth import ( cancel_generation_task, generation_task_status, recover_generation_task, require_alipay_login, render_logged_in_account, render_payment_purchase, start_generation_task, ) from i18n import ( LANGUAGE_OPTIONS, get_amj_csv_template, get_asq_txt_template, get_output_suffix, get_target_flag, get_target_language, tr, ) st.set_page_config( page_title="Article Agent", page_icon="📊", layout="wide", initial_sidebar_state="collapsed", ) WORKSPACE_DIR = APP_CONFIG.get("workspace_dir", ".workspace") os.makedirs(WORKSPACE_DIR, exist_ok=True) os.environ["STREAMLIT_SERVER_ENABLE_CORS"] = "false" os.environ["STREAMLIT_SERVER_ENABLE_XSRF_PROTECTION"] = "false" WECHAT_BLOCKER_JS = """""" components.html(WECHAT_BLOCKER_JS, height=0, width=0) def inject_file_uploader_i18n(language: str) -> None: """Localize Streamlit uploaders without modifying their React structure.""" payload = { "browse": tr(language, "browse_files"), "uploaders": [ { "title": tr(language, "drop_docs"), "limit": tr(language, "limit_docs"), }, { "title": tr(language, "drop_data"), "limit": tr(language, "limit_data"), }, { "title": tr(language, "drop_docs"), "limit": tr(language, "limit_docs"), }, { "title": tr(language, "drop_interviews"), "limit": tr(language, "limit_interviews"), }, ], } script = f""" """ components.html(script, height=0, width=0) def inject_wechat_browser_notice() -> None: """Warn immediately when SmartPaper is opened inside WeChat or WeCom.""" script = """ """ components.html(script, height=0, width=0) for key, default in { "is_running": False, "zip_buffer": None, "zip_filename": "", "run_mode": "amj", "sp_generation_task": None, "sp_pending_launch": None, "sp_pending_resume_task_id": "", "sp_pending_cancel_task_id": "", "sp_payment_required": False, "sp_terminal_notice": None, "sp_task_recovery_initialized": False, "sp_last_ui_task_status": "", }.items(): if key not in st.session_state: st.session_state[key] = default def _safe_key(value: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9_.-]", "_", str(value or "").strip()) return cleaned[:100] or "anonymous" def _new_generation_idempotency_key() -> str: return f"gen_{int(time.time() * 1000)}_{secrets.token_urlsafe(18)}" def _atomic_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8") os.replace(temporary, path) def _canonical_request_hash(value: dict) -> str: material = dict(value or {}) material.pop("request_hash", None) encoded = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _task_dir(task_id: str) -> Path: return Path(_user_workspace) / "tasks" / _safe_key(task_id) def _task_output_path(task: dict) -> Path | None: task_id = str(task.get("task_id") or "").strip() filename = os.path.basename(str(task.get("output_filename") or "").strip()) if not task_id or not filename: return None return _task_dir(task_id) / "outputs" / filename def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]: saved = [] for index, item in enumerate(items or []): if item is None: continue original = os.path.basename(str(getattr(item, "name", f"input_{index}"))) safe_name = re.sub(r"[^A-Za-z0-9_.-]", "_", original)[:140] or f"input_{index}" filename = f"{prefix}_{index:02d}_{safe_name}" path = inputs_dir / filename data = item.getvalue() if hasattr(item, "getvalue") else item.read() path.write_bytes(bytes(data or b"")) saved.append(str(Path("inputs") / filename)) try: item.seek(0) except Exception: pass return saved def _persist_task_request(task: dict, mode: str, lang_code: str) -> dict: task_id = str(task.get("task_id") or "").strip() numeric_id = (st.session_state.get("sp_user") or {}).get("numeric_id") if not task_id or not isinstance(numeric_id, int) or numeric_id <= 0: raise RuntimeError("task_identity_missing") task_dir = _task_dir(task_id) inputs_dir = task_dir / "inputs" inputs_dir.mkdir(parents=True, exist_ok=True) if mode == "amj": refs = _save_uploaded(st.session_state.get("up_amj_ref") or [], inputs_dir, "ref") data = _save_uploaded([st.session_state.get("up_amj_data")], inputs_dir, "data") interviews = [] else: refs = _save_uploaded(st.session_state.get("up_asq_ref") or [], inputs_dir, "ref") data = [] interviews = _save_uploaded([st.session_state.get("up_asq_data")], inputs_dir, "interview") request_data = { "task_id": task_id, "user_id": numeric_id, "mode": mode, "interface_language": lang_code, "target_lang": get_target_language(lang_code), "output_suffix": get_output_suffix(lang_code), "topic": str(st.session_state.get("sp_topic") or DEFAULT_TOPIC), "keywords": str(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS), "target_words": int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]), "reference_files": refs, "data_files": data, "interview_files": interviews, "created_at": datetime.now(timezone.utc).isoformat(), } request_data["request_hash"] = _canonical_request_hash(request_data) _atomic_json(task_dir / "request.json", request_data) return request_data def _pid_alive(pid: int) -> bool: if not isinstance(pid, int) or pid <= 0: return False try: os.kill(pid, 0) return True except OSError: return False def _pid_is_task_executor(pid: int, task_id: str, task_dir: Path) -> bool: """Best-effort Linux identity check before trusting/killing a launch PID. Hugging Face Spaces run on Linux. A stale PID can be reused, so a bare ``os.kill(pid, 0)`` is never enough to identify our executor. """ if not _pid_alive(pid): return False try: raw = (Path("/proc") / str(pid) / "cmdline").read_bytes() command = raw.replace(b"\x00", b" ").decode("utf-8", errors="ignore") except Exception: return False return ( "task_executor.py" in command and f"--task-id {task_id}" in command and str(task_dir.resolve()) in command ) def _launch_executor(task_id: str, *, force: bool = False) -> int: """Start a detached executor without making Streamlit its lifecycle owner. ``subprocess.Popen`` only proves that the OS accepted the child process. R3.2 therefore records the PID and returns immediately. The executor owns claim/heartbeat/checkpoint/finalization; Worker/D1 owns task authority and startup-lease expiry. Streamlit never waits for claim and never terminates an executor because a UI-side timer expired. """ task_dir = _task_dir(task_id) request_path = task_dir / "request.json" if not request_path.is_file(): raise RuntimeError("task_request_manifest_missing") request_data = json.loads(request_path.read_text(encoding="utf-8")) numeric_id = int(request_data.get("user_id") or 0) request_hash = str(request_data.get("request_hash") or "").strip() if numeric_id <= 0 or not request_hash: raise RuntimeError("task_request_identity_invalid") launch_marker_path = task_dir / "executor_launch.json" claim_marker_path = task_dir / "executor_claim.json" if launch_marker_path.is_file(): try: marker = json.loads(launch_marker_path.read_text(encoding="utf-8")) pid = int(marker.get("pid") or 0) if _pid_is_task_executor(pid, task_id, task_dir) and not force: logging.info("[SP_UI] executor_already_spawned task_id=%s pid=%s", task_id, pid) return pid except Exception: pass # A stale local claim marker can never establish ownership for a new # launch/resume. Epoch fencing in Worker/D1, not process termination here, # prevents an older executor from overwriting a newer epoch. try: claim_marker_path.unlink(missing_ok=True) except TypeError: if claim_marker_path.exists(): claim_marker_path.unlink() log_path = task_dir / "executor.log" task_dir.mkdir(parents=True, exist_ok=True) command = [ sys.executable, str(Path(__file__).resolve().parent / "task_executor.py"), "--task-dir", str(task_dir.resolve()), "--task-id", str(task_id), "--user-id", str(numeric_id), ] executor_env = os.environ.copy() executor_env["SMARTPAPER_EXECUTOR_LOG_PATH"] = str(log_path.resolve()) process = subprocess.Popen( command, cwd=str(Path(__file__).resolve().parent), stdin=subprocess.DEVNULL, # Inherit stdout/stderr so executor bootstrap/errors are visible in HF # Container Logs; task_executor also writes the same log to executor.log. close_fds=True, start_new_session=True, env=executor_env, ) _atomic_json( launch_marker_path, { "pid": process.pid, "task_id": str(task_id), "request_hash": request_hash, "launched_at": datetime.now(timezone.utc).isoformat(), }, ) logging.info("[SP_UI] executor_spawned task_id=%s pid=%s", task_id, process.pid) return int(process.pid) def _process_pending_cancel() -> None: task_id = str(st.session_state.get("sp_pending_cancel_task_id") or "").strip() if not task_id: return result = cancel_generation_task(task_id) if result.get("cancelled"): shutil.rmtree(_task_dir(task_id), ignore_errors=True) st.session_state["sp_pending_cancel_task_id"] = "" st.session_state["sp_generation_task"] = dict(result.get("task") or {}) or None st.rerun() def _process_pending_resume() -> None: task_id = str(st.session_state.get("sp_pending_resume_task_id") or "").strip() if not task_id: return if not (_task_dir(task_id) / "request.json").is_file(): st.session_state["sp_pending_resume_task_id"] = "" st.session_state["sp_pending_cancel_task_id"] = task_id st.session_state["sp_terminal_notice"] = { "kind": "error", "message": "Local checkpoint/request files are unavailable after runtime replacement; the reserved quota will be released.", } st.rerun() _launch_executor(task_id, force=True) st.session_state["sp_pending_resume_task_id"] = "" st.session_state["sp_scroll_to_task_monitor"] = True st.rerun() def _process_pending_launch(lang_code: str) -> None: pending = st.session_state.get("sp_pending_launch") if not isinstance(pending, dict): return mode = str(pending.get("mode") or "").strip().lower() idempotency_key = str(pending.get("idempotency_key") or "").strip() try: response = start_generation_task( idempotency_key=idempotency_key, mode=mode, interface_language=lang_code, target_words=int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]), ) except Exception as exc: st.session_state["sp_pending_launch"] = None raw = str(exc or "") if "payment_required" in raw: st.session_state["sp_payment_required"] = True st.error(tr(lang_code, "quota_exhausted")) render_payment_purchase(lang_code) else: st.error(raw) return task = dict(response.get("task") or {}) task_id = str(task.get("task_id") or "").strip() if not task_id: st.session_state["sp_pending_launch"] = None st.error("task_start_missing_task_id") return st.session_state["sp_generation_task"] = task st.session_state["sp_last_ui_task_status"] = str(task.get("status") or "") try: request_path = _task_dir(task_id) / "request.json" if not request_path.is_file(): _persist_task_request(task, mode, lang_code) _launch_executor(task_id) except Exception as exc: st.session_state["sp_pending_launch"] = None st.session_state["sp_pending_cancel_task_id"] = task_id st.session_state["sp_terminal_notice"] = {"kind": "error", "message": str(exc)[:800]} st.rerun() st.session_state["sp_pending_launch"] = None st.session_state["sp_payment_required"] = False st.session_state["sp_scroll_to_task_monitor"] = True st.rerun() inject_wechat_browser_notice() DEFAULT_TOPIC = "Generative Artificial Intelligence in Ubiquitous Learning: Evaluating a Chatbot-based Recommendation Engine for Personalized and Context-aware Education" DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Personalized educational; Educational technologies; HCI; Recommendation system; LLMs; Generative AI; ChatGPT; Prompting engineering; Few-shot prompting" def _current_language_code() -> str: choice = st.session_state.get("interface_language_choice") return LANGUAGE_OPTIONS.get(choice, "zh") authenticated_user = require_alipay_login(_current_language_code()) if not bool(st.session_state.get("sp_task_recovery_initialized")): recover_generation_task() st.session_state["sp_task_recovery_initialized"] = True _user_workspace = os.path.join(WORKSPACE_DIR, "users", _safe_key(authenticated_user.get("id"))) os.makedirs(_user_workspace, exist_ok=True) ACTIVE_TASK_STATUSES = {"created", "running", "resumable", "cancelling"} def _render_monitor_box(task: dict, lang_code: str) -> None: status = str(task.get("status") or "") node = str(task.get("current_node") or status or "") progress = max(0, min(100, int(task.get("progress_pct") or 0))) if status in {"created", "running", "resumable", "cancelling"}: st.progress(progress, text=f"{progress}% — {node}") st.markdown(f"""
{tr(lang_code, 'monitor_pipeline')}: {status}
{tr(lang_code, 'monitor_node')}: {node}
{tr(lang_code, 'monitor_log')}: checkpoint {int(task.get('checkpoint_seq') or 0)} · {str(task.get('checkpoint_stage') or '')}

{tr(lang_code, 'monitor_compute')}:
🧠 DeepSeek: {int(task.get('ds_calls') or 0)} 📚 S2: {int(task.get('s2_calls') or 0)} 🌐 OpenAlex: {int(task.get('oa_calls') or 0)}
""", unsafe_allow_html=True) def _render_task_status_once(lang_code: str) -> None: task = st.session_state.get("sp_generation_task") or {} task_id = str(task.get("task_id") or "").strip() if not task_id: return previous_status = str(st.session_state.get("sp_last_ui_task_status") or str(task.get("status") or "")) status = str(task.get("status") or "") if status in {"created", "running", "resumable", "cancelling"}: try: result = generation_task_status(task_id) task = dict(result.get("task") or task) st.session_state["sp_generation_task"] = task status = str(task.get("status") or "") except Exception as exc: logging.warning("[SP_UI] status_poll_error task_id=%s error=%s", task_id, type(exc).__name__) st.session_state["sp_last_ui_task_status"] = status _render_monitor_box(task, lang_code) if status == "completed": output_path = _task_output_path(task) if output_path and output_path.is_file(): st.success(tr(lang_code, "success")) st.download_button(label=tr(lang_code, "download"), data=output_path.read_bytes(), file_name=output_path.name, mime="application/zip") else: st.warning("The task completed, but the local output file is unavailable after runtime replacement.") elif status == "failed": st.error(str(task.get("safe_error_detail") or task.get("safe_error_code") or tr(lang_code, "node_error"))) elif status == "cancelled": detail = str(task.get("safe_error_detail") or "") if detail: st.warning(detail) # Only boundary transitions need one full app rerun so static controls pick # up resume/terminal state. Running progress never reruns the whole page. if previous_status and status != previous_status and status in {"resumable", "completed", "failed", "cancelled"}: st.rerun() def _scroll_to_task_monitor_once() -> None: """One-time UX correction after launch/resume; never polls or reruns.""" if not bool(st.session_state.get("sp_scroll_to_task_monitor")): return st.session_state["sp_scroll_to_task_monitor"] = False components.html( """ """, height=0, width=0, ) def render_workspace_controls(user): top_toolbar = st.container() with top_toolbar: language_col, account_col = st.columns([5.4, 4.6]) with language_col: lang_choice = st.radio( "Language", list(LANGUAGE_OPTIONS.keys()), index=0, horizontal=True, label_visibility="collapsed", key="interface_language_choice", ) lang_code = LANGUAGE_OPTIONS[lang_choice] st.session_state["sp_lang_code"] = lang_code st.session_state["sp_target_lang"] = get_target_language(lang_code) st.session_state["sp_output_suffix"] = get_output_suffix(lang_code) st.session_state["sp_target_flag"] = get_target_flag(lang_code) with account_col: render_logged_in_account(lang_code, user) task = st.session_state.get("sp_generation_task") or {} task_status = str(task.get("status") or "") active = task_status in ACTIVE_TASK_STATUSES st.session_state["is_running"] = active st.markdown(f"

{tr(lang_code, 'title')}

", unsafe_allow_html=True) if task_status == "resumable": st.warning(tr(lang_code, "checkpoint_warning")) colA, colB = st.columns([1, 1]) if colA.button(tr(lang_code, "resume"), key="resume_task"): st.session_state["sp_pending_resume_task_id"] = str(task.get("task_id") or "") st.rerun() if colB.button(tr(lang_code, "clear"), key="clear_task"): st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "") st.rerun() st.markdown("---") st.text_input(tr(lang_code, "topic_label"), value=DEFAULT_TOPIC, disabled=active, key="sp_topic") st.text_input(tr(lang_code, "keywords_label"), value=DEFAULT_KEYWORDS, disabled=active, key="sp_keywords") st.number_input( tr(lang_code, "words_label"), min_value=int(APP_CONFIG["target_words_min"]), max_value=int(APP_CONFIG["target_words_max"]), value=int(APP_CONFIG["target_words_default"]), step=1000, disabled=active, key="sp_words", ) tab_amj, tab_asq = st.tabs([tr(lang_code, "tab_amj"), tr(lang_code, "tab_asq")]) with tab_amj: st.markdown(tr(lang_code, "amj_config")) st.info(tr(lang_code, "amj_info")) st.file_uploader(tr(lang_code, "upload_reference"), type=["pdf", "doc", "docx"], accept_multiple_files=True, key="up_amj_ref", disabled=active) st.file_uploader(tr(lang_code, "upload_stats"), type=["csv", "xlsx", "xls"], key="up_amj_data", disabled=active) with st.expander(tr(lang_code, "amj_expander")): st.markdown(tr(lang_code, "amj_hint")) template = get_amj_csv_template(lang_code) st.code(template, language="csv") st.download_button(label=tr(lang_code, "download_amj_template"), data=template, file_name="AMJ_Data_Template.csv", mime="text/csv", use_container_width=True) if st.button(tr(lang_code, "start_amj"), disabled=active, key="btn_amj"): st.session_state["run_mode"] = "amj" st.session_state["sp_pending_launch"] = {"mode": "amj", "idempotency_key": _new_generation_idempotency_key()} st.rerun() with tab_asq: st.markdown(tr(lang_code, "asq_config")) st.info(tr(lang_code, "asq_info")) st.file_uploader(tr(lang_code, "upload_reference"), type=["pdf", "doc", "docx"], accept_multiple_files=True, key="up_asq_ref", disabled=active) st.file_uploader(tr(lang_code, "upload_interviews"), type=["txt", "pdf", "doc", "docx"], key="up_asq_data", disabled=active) with st.expander(tr(lang_code, "asq_expander")): st.markdown(tr(lang_code, "asq_hint")) template = get_asq_txt_template(lang_code) st.code(template, language="text") st.download_button(label=tr(lang_code, "download_asq_template"), data=template, file_name="ASQ_Interview_Template.txt", mime="text/plain", use_container_width=True) if st.button(tr(lang_code, "start_asq"), disabled=active, key="btn_asq"): st.session_state["run_mode"] = "asq" st.session_state["sp_pending_launch"] = {"mode": "asq", "idempotency_key": _new_generation_idempotency_key()} st.rerun() inject_file_uploader_i18n(lang_code) render_workspace_controls(authenticated_user) lang_code = str(st.session_state.get("sp_lang_code") or _current_language_code()) _notice = st.session_state.pop("sp_terminal_notice", None) if isinstance(_notice, dict) and _notice.get("message"): (st.error if _notice.get("kind") == "error" else st.warning)(str(_notice.get("message"))) _process_pending_cancel() _process_pending_resume() _process_pending_launch(lang_code) if st.session_state.get("sp_payment_required"): render_payment_purchase(lang_code) # R3.2 UI rollback: keep the task monitor at its pre-R3.1.1 location. # It refreshes only on a normal Streamlit rerun (user action/manual page refresh); # there is no periodic fragment and therefore no 5-second page flicker. _task_for_monitor = st.session_state.get("sp_generation_task") or {} if str(_task_for_monitor.get("task_id") or "").strip(): st.markdown("
", unsafe_allow_html=True) _render_task_status_once(lang_code) _scroll_to_task_monitor_once()