メールファイル( .msg )からのデータベース化、及び、閲覧ページ作成¶
基本構成¶
.msgで保存したメールデータを .json にデータベース化する．
.jsonで保存した情報を streamlit を介してWebブラウザから閲覧する．
メインページは一覧表示 ( app.py )
全文閲覧用に詳細表示ページを作成する ( 01_pageview.py )
コード¶
.mgs からの .json データベース化コード¶
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import os, re, argparse, json, datetime, tqdm
import shutil, logging
from pathlib import Path
from typing import List, Optional
import extract_msg # pip install extract-msg
from dateutil import parser as dtparser # pip install python-dateutil
from pydantic import BaseModel, Field, ValidationError
def suppress_google_genai_noise() -> None:
# この警告は google-genai が logging.warning で出してくることが多い
for name in ("google_genai.types", "google_genai", "google.genai"):
lg = logging.getLogger(name)
lg.setLevel(logging.ERROR) # WARNING を出さない
# どこかで basicConfig されてても確実に沈黙させたいなら propagate を切る
# lg.propagate = False
# -----------------------------
# 1) 出力スキーマ（Pydantic）
# -----------------------------
class MailExtraction(BaseModel):
# 「日付」: ニュースの発表日/出来事の日付を優先。無ければメール日付などから推定して YYYY-MM-DD、無理なら null。
date: Optional[str] = Field(
default=None,
description="ニュースとしての主要日付（YYYY-MM-DD）。不明なら null。"
)
company_names: List[str] = Field(
default_factory=list,
description="会社名（複数可）。例: [\"Novartis\", \"PharmaLogic\"]"
)
radionuclides: List[str] = Field(
default_factory=list,
description="放射性核種（複数可）。例: [\"Ac-225\", \"Lu-177\"]"
)
drug_names: List[str] = Field(
default_factory=list,
description="薬剤/候補薬（複数可）。例: [\"[Ac-225]-RTX-2358\"]"
)
english_news_article: Optional[str] = Field(
default=None,
description="メール本文中の英文ニュース記事（あれば原文を抽出、無ければ null）。"
)
japanese_translation: Optional[str] = Field(
default=None,
description="english_news_article の日本語訳（english_news_article が無いなら null）。"
)
summary: str = Field(
description="メール全体（日本語・英語含む）の要約（日本語）。"
)
impact_opinion: str = Field(
description="このニュースの影響・示唆（日本語、推測は推測とわかるように）。"
)
sender_opinion: Optional[str] = Field(
default=None,
description="送信者が冒頭で述べる所感・コメント（個人名/メール等は書かず、内容のみ）。無ければ null。"
)
title_en: str = Field(
description="英語タイトル。8単語以内。ハイフン連結は不要（Python側で行う）。"
)
# ★追加：日本語タイトル（ファイル名や title_en は変えず、タグとして保持）
title_ja: str = Field(
description="title_en に対応する、わかりやすい日本語タイトル。短く自然な見出し。"
)
# -----------------------------
# 2) .msg 読み取り
# -----------------------------
def read_msg(path: Path) -> dict:
"""
extract-msg で .msg から主要フィールドを抽出（process() は不要）。
"""
msg = extract_msg.openMsg(str(path))
try:
subject = (msg.subject or "").strip()
# plain body（無ければ ""）
body = (msg.body or "").strip()
# plain が空で HTML がある場合のフォールバック（HTMLをそのまま入れる）
if (not body) and getattr(msg, "htmlBodyPrepared", None):
try:
body = msg.htmlBodyPrepared.decode("utf-8", errors="replace")
except Exception:
body = ""
# msg.date は datetime か None（send date）
raw_date = ""
iso_date: Optional[str] = None
if getattr(msg, "date", None):
try:
raw_date = str(msg.date)
iso_date = msg.date.date().isoformat()
except Exception:
raw_date = str(msg.date)
sender = (msg.sender or "").strip()
# 長すぎる本文の安全弁
MAX_CHARS = 80_000
if len(body) > MAX_CHARS:
body = body[:MAX_CHARS] + "\n\n[TRUNCATED]\n"
return {
"file_name": path.name,
"subject": subject,
"body": body,
"email_date_iso": iso_date,
"email_date_raw": raw_date,
"sender_raw": sender,
}
finally:
try:
msg.close()
except Exception:
pass
# -----------------------------
# 3) LLM に投げるプロンプト
# -----------------------------
def build_prompt(mail: dict) -> str:
"""
Structured Output 前提で、スキーマに沿う JSON を返すよう強く指示。
個人情報（個人名・メールアドレス等）は sender_opinion に含めない。
"""
instructions = f"""
あなたは放射性医薬品（Radiopharmaceuticals）のニュースを整理するアシスタントです。
以下の「メール」から情報を抽出し、指定スキーマに厳密に従う JSON だけを返してください（JSON以外は出力しない）。
重要ルール:
- "sender_opinion" には送信者が冒頭で述べる所感・コメントを要約して書く。
ただし個人情報は書かない（個人名、メールアドレス、電話番号、部署内の固有名詞などは削除/一般化）。
例: 「（同僚）」「（送信者）」のように匿名化する。
- "english_news_article" は、メール本文中に「ニュース記事としての英文」がまとまって存在する場合のみ、その英文部分を抽出する。
断片的な英語だけなら null でもよい。
- "japanese_translation" は english_news_article の忠実な日本語訳。english_news_article が null なら null。
- "date" はニュースとしての主要日付（プレスリリース日、治験開始/結果発表日など）を優先して YYYY-MM-DD で記載。
本文から判断できなければ、メール日付（参考: {mail.get("email_date_iso")}）を使ってもよい。無理なら null。
- 不明な項目は、配列は []、文字列は null（ただし必須の summary/impact_opinion/title_en/title_ja は空でなく書く）。
- 会社名/核種/薬剤名は本文に出てくる表記を尊重（例: Ac-225, [Ac-225]-RTX-2358 など）。
- "title_en" は英語で8単語以内の短いタイトル（単語区切りのままでOK）。固有名詞は会社名/薬剤名は可。
例: "Ac-225 FAP radiopharmaceutical enters Phase 1/2"
- ★"title_ja" は title_en に対応する、わかりやすい日本語タイトル（短い見出し）。
直訳よりも自然さを優先してよいが、意味は title_en と整合させる。
- "english_news_article", "japanese_translation" は必ず 1 行の文字列にする（改行はスペースに置換）。
- "english_news_article", "japanese_translation" が長すぎる場合は最大6000文字で切り、末尾に " ...[TRUNCATED]" を付ける。
それでも収まりが悪い/不確実なら null にしてよい。
メール情報:
[SUBJECT]
{mail.get("subject")}
[EMAIL_DATE_RAW]
{mail.get("email_date_raw")}
[BODY]
{mail.get("body")}
""".strip()
return instructions
def _get_text_from_parts(resp) -> str:
"""
google-genai の resp.text を使わず、content.parts の text だけを連結して返す。
non-text part（thought_signature 等）が混ざっても警告を出さない。
"""
texts: list[str] = []
for cand in (getattr(resp, "candidates", None) or []):
content = getattr(cand, "content", None)
parts = getattr(content, "parts", None) or []
for p in parts:
t = getattr(p, "text", None)
if t:
texts.append(t)
return "".join(texts).strip()
def _squash_newlines(s: Optional[str]) -> Optional[str]:
"""CR/LF を除去して 1 行に潰す（空白は適度に正規化）"""
if s is None:
return None
# まず改行コードを統一
s = s.replace("\r\n", "\n").replace("\r", "\n")
# 改行をスペースに
s = re.sub(r"\s*\n\s*", " ", s)
# 連続スペースを1つに
s = re.sub(r"[ \t]{2,}", " ", s)
return s.strip()
# -----------------------------
# 4) JSON抽出の保険（まれにモデルが前後にゴミを付ける場合）
# -----------------------------
def _extract_json_fallback(text: str) -> str:
"""
まずは text 全体を JSON として扱う。
失敗したら、最初の { から最後の } を雑に抜き出す（最後の手段）。
"""
t = (text or "").strip()
if not t:
return t
# そのままJSONの可能性
try:
json.loads(t)
return t
except Exception:
pass
# ```json ... ``` の場合
m = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", t, flags=re.DOTALL | re.IGNORECASE)
if m:
cand = m.group(1).strip()
try:
json.loads(cand)
return cand
except Exception:
pass
# 最初の { から最後の } を抜き出し
i = t.find("{")
j = t.rfind("}")
if i != -1 and j != -1 and i < j:
cand = t[i : j + 1].strip()
try:
json.loads(cand)
return cand
except Exception:
pass
return t
# -----------------------------
# 5) Gemini 呼び出し（Structured Output）
# -----------------------------
def extract_with_gemini(
prompt: str,
model: str,
api_key: Optional[str] = None,
temperature: float = 0.0,
) -> MailExtraction:
"""
google-genai を使って JSON schema に沿う応答を生成し、Pydanticで検証。
"""
try:
from google import genai # pip install google-genai
except Exception as e:
raise RuntimeError(
"google-genai が import できません。Geminiモードを使う場合は `pip install google-genai` してください。"
) from e
if api_key:
client = genai.Client(api_key=api_key)
else:
# 環境変数 GEMINI_API_KEY / GOOGLE_API_KEY から読む
client = genai.Client()
schema = MailExtraction.model_json_schema()
resp = client.models.generate_content(
model=model,
contents=prompt,
config={
"temperature": float(temperature),
"response_mime_type": "application/json",
"response_json_schema": schema,
},
)
raw = _get_text_from_parts(resp) # resp.text -> function to load text
raw = _extract_json_fallback(raw)
try:
data = json.loads(raw)
except json.JSONDecodeError as e:
raise RuntimeError(
"Gemini returned invalid JSON (possibly truncated). "
f"JSONDecodeError: {e}\n\nRaw (head):\n{raw[:2000]}\n\nRaw (tail):\n{raw[-2000:]}"
) from e
# ★改行つぶし
data["english_news_article"] = _squash_newlines(data.get("english_news_article"))
data["japanese_translation"] = _squash_newlines(data.get("japanese_translation"))
try:
return MailExtraction.model_validate(data)
except ValidationError as e:
raise RuntimeError(
"Gemini response did not match schema after sanitizing.\n"
f"Validation error:\n{e}\n\nData(head):\n{json.dumps(data, ensure_ascii=False)[:2000]}"
) from e
# -----------------------------
# 6) Ollama 呼び出し（ローカル / Ollama Cloudでも可）
# -----------------------------
def extract_with_ollama(
prompt: str,
model: str,
host: Optional[str] = None,
temperature: float = 0.0,
) -> MailExtraction:
"""
Ollamaの structured outputs を使って JSON schema に沿う応答を生成し、Pydanticで検証。
- ローカル: Ollamaアプリ起動 + `ollama pull <model>` 済みで利用
- host を指定するとリモートOllamaにも接続可能（例: http://localhost:11434）
"""
try:
import ollama # pip install ollama
from ollama import Client
except Exception as e:
raise RuntimeError(
"ollama Python が import できません。Ollamaモードを使う場合は `pip install ollama` してください。"
) from e
schema = MailExtraction.model_json_schema()
if host:
client = Client(host=host)
resp = client.chat(
model=model,
messages=[{"role": "user", "content": prompt}],
format=schema, # ★ JSON Schemaを強制
options={"temperature": float(temperature)},
stream=False,
)
raw = resp.message.content
else:
resp = ollama.chat(
model=model,
messages=[{"role": "user", "content": prompt}],
format=schema, # ★ JSON Schemaを強制
options={"temperature": float(temperature)},
stream=False,
)
raw = resp.message.content if hasattr(resp, "message") else resp["message"]["content"]
raw = _extract_json_fallback(raw)
try:
return MailExtraction.model_validate_json(raw)
except ValidationError as e:
raise RuntimeError(
"Ollama response did not match schema. "
f"Raw response text:\n{raw}\n\nValidation error:\n{e}"
)
# -----------------------------
# 7) 入出力（単体/ディレクトリ対応）
# -----------------------------
def iter_msg_files(input_path: Path) -> List[Path]:
if input_path.is_file():
return [input_path]
processed_dir = input_path / "processed"
files: List[Path] = []
for p in input_path.rglob("*.msg"):
if not p.is_file():
continue
# processed/ 配下は除外（再処理防止）
try:
if p.is_relative_to(processed_dir):
continue
except AttributeError:
# Python < 3.9
try:
p.relative_to(processed_dir)
continue
except ValueError:
pass
files.append(p)
return sorted(files)
def ensure_out_path(out: Path, input_is_dir: bool) -> None:
if input_is_dir:
out.mkdir(parents=True, exist_ok=True)
else:
out.parent.mkdir(parents=True, exist_ok=True)
def _yyyymmdd_from_dates(primary_iso: str | None, fallback_iso: str | None) -> str:
"""
YYYYMMDD を作る。primary_iso 優先、無ければ fallback_iso、無ければ今日。
"""
for iso in (primary_iso, fallback_iso):
if iso:
try:
d = dtparser.parse(iso).date()
return d.strftime("%Y%m%d")
except Exception:
pass
return datetime.date.today().strftime("%Y%m%d")
def _slug_from_title_en(title_en: str) -> str:
"""
英語タイトル（単語区切り）→ 8単語に制限 → '-' 連結 → ファイル名に安全な slug 化
"""
words = re.findall(r"[A-Za-z0-9\[\]\(\)\+\.]+(?:-[A-Za-z0-9\[\]\(\)\+\.]+)?", title_en)
words = words[:8] if words else []
slug = "-".join(words).strip("-").lower()
slug = re.sub(r"[^a-z0-9\-\[\]\(\)\+\.]+", "-", slug)
slug = re.sub(r"-{2,}", "-", slug).strip("-")
return slug or "news-update"
def build_final_title(result_date_iso: str | None, email_date_iso: str | None, title_en: str) -> str:
yyyymmdd = _yyyymmdd_from_dates(result_date_iso, email_date_iso)
slug = _slug_from_title_en(title_en)
return f"{yyyymmdd}-{slug}"
def unique_path(base: Path) -> Path:
"""
既に存在する場合は -2, -3... を付けてユニークにする
"""
if not base.exists():
return base
stem = base.stem
suffix = base.suffix
parent = base.parent
i = 2
while True:
cand = parent / f"{stem}-{i}{suffix}"
if not cand.exists():
return cand
i += 1
def move_to_bucket(msg_file: Path, in_path: Path, input_is_dir: bool, bucket: str) -> Path:
"""
msg_file を processed_<bucket>/ 配下に移動する。
bucket 例: "ok", "failed"
"""
if input_is_dir:
root = in_path / f"processed_{bucket}"
rel = msg_file.relative_to(in_path)
dst = root / rel
else:
root = in_path.parent / f"processed_{bucket}"
dst = root / msg_file.name
root.mkdir(parents=True, exist_ok=True)
dst.parent.mkdir(parents=True, exist_ok=True)
dst = unique_path(dst)
shutil.move(str(msg_file), str(dst))
return dst
def main():
ap = argparse.ArgumentParser()
ap.add_argument(
"-i", "--input", type=str, default="msg",
help="入力 .msg ファイル、または .msg を含むディレクトリ"
)
ap.add_argument(
"-o", "--output", type=str, default="json",
help="出力先（省略時: ./json）。単体入力なら親ディレクトリとして扱い title.json を作成"
)
# ★追加：バックエンド切替
ap.add_argument(
"--backend", type=str, default="gemini", choices=["gemini", "ollama"],
help="使用するLLMバックエンド（gemini: API / ollama: ローカルLLM）"
)
# Gemini用
ap.add_argument(
"--model", type=str, default="gemini-flash-latest",
help="Gemini model name (default: gemini-flash-latest)"
)
ap.add_argument(
"--api-key", type=str, default=None,
help="Gemini APIキーを直接渡す（通常は環境変数 GEMINI_API_KEY 推奨）"
)
# Ollama用
ap.add_argument(
"--ollama-model", type=str, default="gemma3",
help="Ollama model name (default: gemma3). 例: gemma3, llama3.2, qwen2.5, etc."
)
ap.add_argument(
"--ollama-host", type=str, default=None,
help="Ollama host (例: http://localhost:11434). 未指定なら環境変数 OLLAMA_HOST / デフォルトを使用"
)
# 共通
ap.add_argument(
"--temperature", type=float, default=0.0,
help="生成温度（structured outputでは 0 推奨）"
)
ap.add_argument(
"--dry-run", action="store_true",
help="LLMには投げず、抽出対象テキスト確認のみ行う"
)
args = ap.parse_args()
in_path = Path(args.input).expanduser().resolve()
out_path = Path(args.output).expanduser().resolve()
msg_files = iter_msg_files(in_path)
if not msg_files:
raise SystemExit(f"No .msg files found: {in_path}")
suppress_google_genai_noise()
input_is_dir = in_path.is_dir()
ensure_out_path(out_path, input_is_dir=input_is_dir)
# ログファイルは out_path の解釈に合わせる（out_pathがdirならそこ、fileなら親）
log_dir = out_path if (input_is_dir or out_path.is_dir() or str(out_path).endswith(os.sep)) else out_path.parent
log_dir.mkdir(parents=True, exist_ok=True)
log_file = log_dir / "msg2json.log"
logger = logging.getLogger("msg2json")
logger.setLevel(logging.INFO)
logger.handlers.clear()
fh = logging.FileHandler(log_file, encoding="utf-8")
fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
logger.addHandler(fh)
logger.propagate = False
ok_count = 0
fail_count = 0
for msg_file in tqdm.tqdm(msg_files):
try:
mail = read_msg(msg_file)
if args.dry_run:
print(f"--- {msg_file} ---")
print("SUBJECT:", mail["subject"])
print("EMAIL_DATE_RAW:", mail["email_date_raw"])
print("BODY(head):", mail["body"][:500])
continue
prompt = build_prompt(mail)
# ===== LLM 呼び出し（ここが一番落ちる）=====
try:
if args.backend == "gemini":
result = extract_with_gemini(
prompt,
model=args.model,
api_key=args.api_key,
temperature=args.temperature,
)
else:
result = extract_with_ollama(
prompt,
model=args.ollama_model,
host=args.ollama_host,
temperature=args.temperature,
)
except Exception as e:
fail_count += 1
logger.exception(f"[FAILED] LLM parse error: {msg_file} : {e}")
# 失敗した msg は再処理を避けるため退避
try:
moved = move_to_bucket(msg_file, in_path, input_is_dir, bucket="failed")
logger.info(f"[FAILED] Moved: {msg_file} -> {moved}")
except Exception as me:
logger.exception(f"[FAILED] Could not move failed msg: {msg_file} : {me}")
continue # ★次へ
# ===== ここから先は「結果の保存・移動」=====
payload = result.model_dump()
final_title = build_final_title(
result_date_iso=payload.get("date"),
email_date_iso=mail.get("email_date_iso"),
title_en=payload.get("title_en", ""),
)
payload["title"] = final_title
if input_is_dir or out_path.is_dir() or str(out_path).endswith(os.sep):
out_dir = out_path
out_dir.mkdir(parents=True, exist_ok=True)
else:
out_dir = out_path.parent
out_dir.mkdir(parents=True, exist_ok=True)
dst = unique_path(out_dir / f"{final_title}.json")
dst.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
logger.info(f"[OK] Wrote: {dst}")
# 成功した msg を processed_ok へ
try:
moved = move_to_bucket(msg_file, in_path, input_is_dir, bucket="ok")
logger.info(f"[OK] Moved: {msg_file} -> {moved}")
except Exception as me:
# JSONは書けたが msg 移動に失敗、という状態なので止めずにログだけ
logger.exception(f"[OK] Could not move msg (json already written): {msg_file} : {me}")
ok_count += 1
except KeyboardInterrupt:
raise
except Exception as e:
# read_msg など「想定外」で落ちた場合もスキップして続行
fail_count += 1
logger.exception(f"[FAILED] Unexpected error: {msg_file} : {e}")
try:
moved = move_to_bucket(msg_file, in_path, input_is_dir, bucket="failed")
logger.info(f"[FAILED] Moved: {msg_file} -> {moved}")
except Exception as me:
logger.exception(f"[FAILED] Could not move failed msg: {msg_file} : {me}")
continue
logger.info(f"Done. ok={ok_count}, failed={fail_count}, total={len(msg_files)}")
# for msg_file in tqdm.tqdm( msg_files ):
# mail = read_msg(msg_file)
# if args.dry_run:
# print(f"--- {msg_file} ---")
# print("SUBJECT:", mail["subject"])
# print("EMAIL_DATE_RAW:", mail["email_date_raw"])
# print("BODY(head):", mail["body"][:500])
# continue
# prompt = build_prompt(mail)
# # ★ここでバックエンド分岐
# if args.backend == "gemini":
# result = extract_with_gemini(
# prompt,
# model=args.model,
# api_key=args.api_key,
# temperature=args.temperature,
# )
# else:
# result = extract_with_ollama(
# prompt,
# model=args.ollama_model,
# host=args.ollama_host,
# temperature=args.temperature,
# )
# payload = result.model_dump()
# # title を最終生成（YYYYMMDD- + 英語8単語以内を'-'連結）
# final_title = build_final_title(
# result_date_iso=payload.get("date"),
# email_date_iso=mail.get("email_date_iso"),
# title_en=payload.get("title_en", ""),
# )
# payload["title"] = final_title # ← JSON に含める（要件）
# # 出力先の解釈：
# # - 入力がディレクトリなら out_path はディレクトリ扱い
# # - 入力が単体でも、out_path がディレクトリならそこへ title.json
# # - out_path がファイルパスでも、要件に従い parent に title.json を作る
# if input_is_dir or out_path.is_dir() or str(out_path).endswith(os.sep):
# out_dir = out_path
# out_dir.mkdir(parents=True, exist_ok=True)
# else:
# out_dir = out_path.parent
# out_dir.mkdir(parents=True, exist_ok=True)
# dst = out_dir / f"{final_title}.json"
# dst = unique_path(dst)
# dst.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# # ★標準出力ではなくログへ
# logger.info(f"Wrote: {dst}")
# # ★処理済みファイルを processed/ へ移動（同じファイルが何度も対象になるのを防ぐ）
# if input_is_dir:
# processed_root = in_path / "processed"
# rel = msg_file.relative_to(in_path) # 元の相対パスを維持
# moved_dst = processed_root / rel # processed/<元の相対パス>
# else:
# processed_root = in_path.parent / "processed"
# moved_dst = processed_root / msg_file.name
# processed_root.mkdir(parents=True, exist_ok=True)
# moved_dst.parent.mkdir(parents=True, exist_ok=True)
# moved_dst = unique_path(moved_dst) # 同名衝突回避
# shutil.move(str(msg_file), str(moved_dst))
# logger.info(f"Moved: {msg_file} -> {moved_dst}")
if __name__ == "__main__":
main()
streamlit 閲覧ページ生成コード¶
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import html
import json
from pathlib import Path
from typing import Any, Dict, List, Optional
import pandas as pd
import streamlit as st
# ★ ステータス（記事ページ側と揃える）
ALLOWED_STATUS = ["New", "既読", "要再確認"]
# 目次表でのプレビュー量（文字数）
PREVIEW_MAX_CHARS = 2600
def safe_load_json(p: Path) -> Optional[Dict[str, Any]]:
try:
return json.loads(p.read_text(encoding="utf-8"))
except Exception:
return None
def ensure_user_fields(d: Dict[str, Any]) -> Dict[str, Any]:
# 追記欄の既定値だけ保証
if "user_comment" not in d:
d["user_comment"] = ""
if "user_rating" not in d:
d["user_rating"] = None
if "user_status" not in d:
d["user_status"] = "New"
if "user_updated_at" not in d:
d["user_updated_at"] = None
# 既存データ互換（旧: new/reviewed/shared/ignored → 新: New/既読/要再確認）
old = str(d.get("user_status") or "")
mapping = {
"new": "New",
"reviewed": "既読",
"shared": "既読",
"ignored": "要再確認",
}
if old in mapping:
d["user_status"] = mapping[old]
if d.get("user_status") not in ALLOWED_STATUS:
d["user_status"] = "New"
# ratingをintに正規化
r = d.get("user_rating")
if r is not None:
try:
r = int(r)
d["user_rating"] = r if 1 <= r <= 5 else None
except Exception:
d["user_rating"] = None
return d
def join_list(x: Any) -> str:
if not x:
return ""
if isinstance(x, list):
return ", ".join(str(i) for i in x)
return str(x)
def normalize_text(x: Any) -> str:
if not x:
return ""
return str(x).replace("\r\n", "\n").replace("\r", "\n")
def preview_text(x: Any) -> str:
t = " ".join(normalize_text(x).split())
if len(t) > PREVIEW_MAX_CHARS:
t = t[:PREVIEW_MAX_CHARS] + "…"
return t
def title_of(d: Dict[str, Any]) -> str:
return str(d.get("title") or d.get("subject") or d.get("file_name") or "untitled")
def parse_date_for_sort(d: Dict[str, Any]) -> pd.Timestamp:
for k in ["date", "email_date_iso"]:
v = d.get(k)
if v:
try:
return pd.to_datetime(v, errors="raise")
except Exception:
pass
return pd.Timestamp.min
def rating_to_stars(r: Any) -> str:
"""目次表示専用：★のみ（☆☆なし）"""
if r is None or (isinstance(r, float) and pd.isna(r)):
return ""
try:
rr = int(r)
except Exception:
return ""
rr = max(0, min(5, rr))
return "★" * rr if rr > 0 else ""
def list_json_files(root: Path) -> List[Path]:
return sorted([p for p in root.rglob("*.json") if p.is_file()])
@st.cache_data(show_spinner=False)
def load_index(json_dir: str) -> pd.DataFrame:
root = Path(json_dir).expanduser().resolve()
rows = []
for p in list_json_files(root):
d = safe_load_json(p)
if not isinstance(d, dict):
continue
d = ensure_user_fields(d)
rows.append({
"_path": str(p),
"_sort_date": parse_date_for_sort(d),
# 並び順は後でcolsで制御するので、ここは全項目を持つ
"No.": None, # 後で採番
"日付": d.get("date") or "",
"タイトル": title_of(d),
"リンク": "", # HTML生成時に作る（ここはダミー）
"状態": d.get("user_status") or "New",
"重要度": rating_to_stars(d.get("user_rating")),
"要約": preview_text(d.get("summary")),
"会社": join_list(d.get("company_names")),
"核種": join_list(d.get("radionuclides")),
"薬剤": join_list(d.get("drug_names")),
"コメント": preview_text(d.get("user_comment")),
"更新日時": d.get("user_updated_at") or "",
"AI影響評価": preview_text(d.get("impact_opinion")),
"英文": preview_text(d.get("english_news_article")),
"和訳": preview_text(d.get("japanese_translation")),
"他者コメント": preview_text(d.get("sender_opinion")),
})
df = pd.DataFrame(rows)
if len(df) == 0:
return df
df = df.sort_values(by=["_sort_date", "タイトル"], ascending=[False, True], kind="stable").reset_index(drop=True)
df["No."] = range(1, len(df) + 1)
return df
def render_html_table(df: pd.DataFrame, clamp_lines: int) -> None:
st.markdown(f"""
<style>
.table-wrap {{
overflow-x: auto;
padding-bottom: 8px;
}}
.news-table {{
border-collapse: separate;
border-spacing: 0 10px;
width: max-content;
font-size: 0.80rem;
line-height: 1.20;
}}
.news-table tbody tr {{
background: rgba(255,255,255,0.03);
outline: 1px solid rgba(255,255,255,0.08);
border-radius: 10px;
}}
.news-table td, .news-table th {{
padding: 8px 10px;
vertical-align: top;
border: none;
}}
.news-table thead th {{
position: sticky;
top: 0;
background: rgba(0,0,0,0.60);
z-index: 5;
outline: 1px solid rgba(255,255,255,0.12);
border-radius: 8px;
}}
.clamp {{
display: -webkit-box;
-webkit-line-clamp: {clamp_lines};
-webkit-box-orient: vertical;
overflow: hidden;
white-space: normal;
}}
.title {{
font-weight: 400;
white-space: normal;
}}
a {{
text-decoration: none;
}}
a:hover {{
text-decoration: underline;
}}
.goto a {{
display: inline-block;
padding: 4px 8px;
border-radius: 8px;
border: 1px solid rgba(255,255,255,0.18);
background: rgba(100,200,255,0.10);
}}
.goto a:hover {{
background: rgba(100,200,255,0.18);
}}
/* 列幅（メタ列は細く） */
.w-no {{ min-width: 46px; max-width: 60px; }}
.w-date {{ min-width: 80px; max-width: 110px; }}
.w-title {{ min-width: 260px; max-width: 520px; }}
.w-link {{ min-width: 70px; max-width: 90px; }}
.w-state {{ min-width: 70px; max-width: 110px; }}
.w-rate {{ min-width: 70px; max-width: 110px; }}
.w-summary{{ min-width: 520px; max-width: 860px; }}
.w-company{{ min-width: 120px; max-width: 200px; }}
.w-nuclide{{ min-width: 90px; max-width: 140px; }}
.w-drug {{ min-width: 120px; max-width: 200px; }}
.w-comment{{ min-width: 360px; max-width: 620px; }}
.w-upd {{ min-width: 140px; max-width: 220px; }}
.w-impact {{ min-width: 520px; max-width: 860px; }}
.w-en {{ min-width: 520px; max-width: 860px; }}
.w-ja {{ min-width: 520px; max-width: 860px; }}
.w-other {{ min-width: 520px; max-width: 860px; }}
</style>
""", unsafe_allow_html=True)
# ★ 指定の列順
cols = [
("No.", "w-no"),
("日付", "w-date"),
("タイトル", "w-title"),
("リンク", "w-link"),
("状態", "w-state"),
("重要度", "w-rate"),
("要約", "w-summary"),
("会社", "w-company"),
("核種", "w-nuclide"),
("薬剤", "w-drug"),
("コメント", "w-comment"),
("更新日時", "w-upd"),
("AI影響評価", "w-impact"),
("英文", "w-en"),
("和訳", "w-ja"),
("他者コメント", "w-other"),
]
thead = "<tr>" + "".join(
f"<th class='{cls}'>{html.escape(name)}</th>" for name, cls in cols
) + "</tr>"
def cell(text: Any, cls: str, clamp: bool = True, extra: str = "") -> str:
s = "" if pd.isna(text) else str(text)
s = html.escape(s)
c = "clamp " if clamp else ""
return f"<td class='{cls}'><div class='{c}{extra}'>{s}</div></td>"
rows_html = []
for _, r in df.iterrows():
p = str(r["_path"])
href = f"/pageview?id={html.escape(p)}"
row = "<tr>"
row += cell(r["No."], "w-no", clamp=False)
row += cell(r["日付"], "w-date", clamp=False)
row += cell(r["タイトル"], "w-title", clamp=True, extra="title")
# リンク列（表示文字は「リンク」固定）
row += f"<td class='w-link goto'><a href='{href}' target='_self'>リンク</a></td>"
row += cell(r["状態"], "w-state", clamp=False)
row += cell(r["重要度"], "w-rate", clamp=False)
row += cell(r["要約"], "w-summary", clamp=True)
row += cell(r["会社"], "w-company", clamp=True)
row += cell(r["核種"], "w-nuclide", clamp=True)
row += cell(r["薬剤"], "w-drug", clamp=True)
row += cell(r["コメント"], "w-comment", clamp=True)
row += cell(r["更新日時"], "w-upd", clamp=False)
row += cell(r["AI影響評価"], "w-impact", clamp=True)
row += cell(r["英文"], "w-en", clamp=True)
row += cell(r["和訳"], "w-ja", clamp=True)
row += cell(r["他者コメント"], "w-other", clamp=True)
row += "</tr>"
rows_html.append(row)
table_html = f"""
<div class="table-wrap">
<table class="news-table">
<thead>{thead}</thead>
<tbody>{''.join(rows_html)}</tbody>
</table>
</div>
"""
st.markdown(table_html, unsafe_allow_html=True)
# ========= UI =========
st.set_page_config(page_title="核医学の情報収集ログ", layout="wide")
st.title("核医学の情報収集ログ")
with st.sidebar:
json_dir = st.text_input("JSONフォルダ（out/など）", value="json")
if st.button("再読み込み", use_container_width=True):
load_index.clear()
st.divider()
st.subheader("フィルタ")
q = st.text_input("検索（タイトル/会社/核種/薬剤/要約/コメント/AI影響/英文/和訳/他者コメント）", value="")
status_filter = st.multiselect("状態（user_status）", options=ALLOWED_STATUS, default=[])
# 重要度は★表示でフィルタ
rating_filter = st.multiselect("重要度（★）", options=[1, 2, 3, 4, 5], default=[])
st.divider()
st.subheader("表示")
clamp_lines = st.slider("本文プレビュー行数（目次）", min_value=3, max_value=14, value=8, step=1)
df = load_index(json_dir)
if df is None or len(df) == 0:
st.warning("JSONが見つかりません。フォルダ指定を確認してください。")
st.stop()
def match_row(r: pd.Series) -> bool:
if status_filter and r["状態"] not in status_filter:
return False
if rating_filter:
# df側は ★文字列なので、その数で判定
star_n = str(r.get("重要度", "")).count("★")
if star_n not in rating_filter:
return False
if q.strip():
qq = q.strip().lower()
hay = " ".join([
str(r.get("タイトル", "")),
str(r.get("会社", "")),
str(r.get("核種", "")),
str(r.get("薬剤", "")),
str(r.get("要約", "")),
str(r.get("コメント", "")),
str(r.get("状態", "")),
str(r.get("重要度", "")),
str(r.get("AI影響評価", "")),
str(r.get("英文", "")),
str(r.get("和訳", "")),
str(r.get("他者コメント", "")),
]).lower()
if qq not in hay:
return False
return True
df_view = df[df.apply(match_row, axis=1)].copy()
st.caption(f"表示: {len(df_view)} / 全体: {len(df)}")
# 全件表示（ページングなし）
if len(df_view) == 0:
st.info("フィルタ結果が0件です。")
st.stop()
render_html_table(df_view, clamp_lines=clamp_lines)
# --- ページングあり --- #
# total = len(df_view)
# if total == 0:
# st.info("フィルタ結果が0件です。")
# st.stop()
# max_page = (total - 1) // rows_per_page + 1
# page = st.number_input("ページ", min_value=1, max_value=max_page, value=1, step=1)
# start = (page - 1) * rows_per_page
# end = min(start + rows_per_page, total)
# df_page = df_view.iloc[start:end].copy()
# render_html_table(df_page, clamp_lines=clamp_lines)
streamlit 詳細ページ用 (部品) コード¶
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import json
import shutil
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Optional
import streamlit as st
# ステータス（指定の3種）
ALLOWED_STATUS = ["New", "既読", "要再確認"]
DEFAULTS = {
"user_comment": "",
"user_rating": None, # 1-5 or None（JSON保存はint）
"user_status": "New",
"user_updated_at": None,
}
def now_iso_jst() -> str:
return datetime.now().strftime("%Y-%m-%dT%H:%M:%S+09:00")
def safe_load_json(p: Path) -> Optional[Dict[str, Any]]:
try:
return json.loads(p.read_text(encoding="utf-8"))
except Exception:
return None
def ensure_user_fields(d: Dict[str, Any]) -> Dict[str, Any]:
for k, v in DEFAULTS.items():
if k not in d:
d[k] = v
if d.get("user_status") not in ALLOWED_STATUS:
d["user_status"] = "New"
r = d.get("user_rating")
if r is not None:
try:
r = int(r)
d["user_rating"] = r if 1 <= r <= 5 else None
except Exception:
d["user_rating"] = None
return d
def join_list(x: Any) -> str:
if not x:
return ""
if isinstance(x, list):
return ", ".join(str(i) for i in x)
return str(x)
def title_of(d: Dict[str, Any]) -> str:
return str(d.get("title") or d.get("subject") or d.get("file_name") or "untitled")
def rating_to_stars(r: Optional[int]) -> str:
"""表示専用：★のみ（空き☆は出さない）"""
if r is None:
return "—"
try:
r = int(r)
except Exception:
return "—"
r = max(0, min(5, r))
return "★" * r if r > 0 else "—"
def stars_to_int(s: str) -> Optional[int]:
"""UI選択（★表記）→ intへ"""
if s == "(なし)":
return None
return s.count("★")
def get_app_root() -> Path:
"""
app.py と同階層を “home” とみなす。
pages/01_pageview.py を想定して、親の親がルート。
"""
here = Path(__file__).resolve()
return here.parents[1]
def backup_file(src: Path) -> Path:
backup_dir = get_app_root() / "backup"
backup_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
dst = backup_dir / f"{src.stem}__{ts}{src.suffix}"
shutil.copy2(src, dst)
return dst
def save_record(path: Path, d: Dict[str, Any]) -> None:
backup_file(path)
d["user_updated_at"] = now_iso_jst()
path.write_text(json.dumps(d, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def get_id_param() -> Optional[str]:
if hasattr(st, "query_params"):
v = st.query_params.get("id")
return v if isinstance(v, str) else (v[0] if v else None)
qp = st.experimental_get_query_params()
return qp.get("id", [None])[0]
# ========= UI =========
st.set_page_config(page_title="核医学の情報収集ログ - 記事", layout="wide")
st.title("核医学の情報収集ログ")
st.page_link("app.py", label="← 目次に戻る")
id_path = get_id_param()
if not id_path:
st.warning("記事ID（id）がありません。目次から開いてください。")
st.stop()
path = Path(id_path)
rec = safe_load_json(path)
if not isinstance(rec, dict):
st.error(f"JSONを読めません: {id_path}")
st.stop()
rec = ensure_user_fields(rec)
st.markdown(f"## {title_of(rec)}")
st.write(f"**日付:** {rec.get('date') or '-'}")
st.write(f"**会社:** {join_list(rec.get('company_names')) or '-'}")
st.write(f"**核種:** {join_list(rec.get('radionuclides')) or '-'}")
st.write(f"**薬剤:** {join_list(rec.get('drug_names')) or '-'}")
st.write(f"**状態:** {rec.get('user_status') or 'New'}")
st.write(f"**重要度:** {rating_to_stars(rec.get('user_rating'))}")
st.caption(f"JSON: {id_path}")
with st.expander("要約", expanded=True):
st.write(rec.get("summary") or "")
with st.expander("AI影響コメント", expanded=True):
st.write(rec.get("impact_opinion") or "")
if rec.get("sender_opinion"):
with st.expander("他者コメント", expanded=False):
st.write(rec.get("sender_opinion") or "")
with st.expander("英文", expanded=False):
st.write(rec.get("english_news_article") or "")
with st.expander("和訳", expanded=False):
st.write(rec.get("japanese_translation") or "")
st.subheader("評価メモ")
with st.form("edit_form", clear_on_submit=False):
c1, c2, c3 = st.columns([0.34, 0.22, 0.44])
with c1:
user_status = st.selectbox(
"状態",
options=ALLOWED_STATUS,
index=ALLOWED_STATUS.index(rec.get("user_status") or "New"),
)
with c2:
# UIは★のみ。保存はint。
rating_options = ["(なし)"] + ["★" * i for i in range(1, 6)]
cur = rec.get("user_rating")
idx = 0 if cur is None else max(1, min(5, int(cur)))
rating_sel = st.selectbox("重要度", options=rating_options, index=idx)
with c3:
st.text_input("更新日時", value=rec.get("user_updated_at") or "-", disabled=True)
user_comment = st.text_area("コメント", value=rec.get("user_comment") or "", height=180)
submitted = st.form_submit_button("保存(.jsonに追記)")
if submitted:
rec["user_status"] = user_status
rec["user_rating"] = stars_to_int(rating_sel) # int/Noneで保存
rec["user_comment"] = user_comment or ""
save_record(path, rec)
st.cache_data.clear() # ★ 目次の load_index キャッシュを無効化
st.success("Saved.")
実行用 tasks.py¶
# tasks.py
# Invoke tasks for msg2json
#
# Usage examples:
# invoke msg2json
# invoke msg2json --input="msg" --output="json"
# invoke msg2json --input="some/file.msg" --output="json" --dry-run
# invoke msg2json --model="gemini-flash-latest"
# invoke msg2json --api-key="YOUR_KEY"
#
# Notes:
# - This assumes your converter script is named "msg2json.py" in the same directory.
# If your filename is different, pass --script="yourname.py" or change DEFAULT_SCRIPT below.
from __future__ import annotations
import os
import shlex
from invoke import task
DEFAULT_SCRIPT = "pyt/msg2json.py"
DEFAULT_APP = "app.py"
def _q(s: str) -> str:
"""Shell-quote helper."""
return shlex.quote(s)
@task(
help={
"input": "Input .msg file or directory (default: msg)",
"output": "Output directory or path (default: json)",
"model": "Gemini model name (default: gemini-flash-latest)",
"api_key": "API key (optional; otherwise rely on env GEMINI_API_KEY/GOOGLE_API_KEY)",
"dry_run": "If set, do not call Gemini; just print extracted mail head/body preview",
"script": "Python script filename to run (default: msg2json.py)",
"python": "Python executable to use (default: python3 or $PYTHON)",
"extra": "Extra raw CLI args appended to the command (optional)",
}
)
def msg2json(
c,
input: str = "msg",
output: str = "json",
model: str = "gemini-flash-latest",
api_key: str | None = None,
dry_run: bool = False,
script: str = DEFAULT_SCRIPT,
python: str | None = None,
extra: str | None = None,
):
"""
Run msg2json converter via Invoke.
This calls:
python msg2json.py -i <input> -o <output> --model <model> [--api-key ...] [--dry-run]
"""
py = python or os.environ.get("PYTHON") or "python3"
cmd_parts = [
_q(py),
_q(script),
"-i",
_q(input),
"-o",
_q(output),
"--model",
_q(model),
]
if api_key:
cmd_parts += ["--api-key", _q(api_key)]
if dry_run:
cmd_parts.append("--dry-run")
if extra:
# Append as-is (user responsibility). If you want safe splitting, pass it like:
# invoke msg2json --extra='--dry-run'
cmd_parts.append(extra)
cmd = " ".join(cmd_parts)
c.run(cmd, pty=True)
@task(
help={
"app": "Streamlit app file (default: app.py)",
"port": "Server port (default: 8501)",
"server_headless": "Run Streamlit in headless mode (useful on servers/CI)",
"browser": "Open browser automatically (default: False; use --browser to enable)",
"extra": "Extra raw args appended to streamlit command (optional)",
"python": "Python executable to use (default: python3 or $PYTHON)",
}
)
def run(
c,
app: str = DEFAULT_APP,
port: int = 8501,
server_headless: bool = False,
browser: bool = False,
extra: str | None = None,
python: str | None = None,
):
"""
Launch Streamlit app:
streamlit run app.py
"""
py = python or os.environ.get("PYTHON") or "python3"
# Use `python -m streamlit` to avoid PATH issues where `streamlit` command isn't found.
cmd_parts = [
_q(py),
"-m",
"streamlit",
"run",
_q(app),
"--server.port",
_q(str(port)),
]
if server_headless:
cmd_parts += ["--server.headless", "true"]
# Streamlit default is to open browser; many people want the opposite in dev shells.
# We treat --browser as an opt-in to open.
if not browser:
cmd_parts += ["--server.headless", "true"] if server_headless else []
# If you want to *force* no browser open locally, Streamlit doesn't have a perfect flag,
# but headless=true is the common approach.
if extra:
cmd_parts.append(extra)
cmd = " ".join(cmd_parts)
c.run(cmd, pty=True)
@task
def version(c, script: str = DEFAULT_SCRIPT, python: str | None = None):
"""Show msg2json.py --help (quick sanity check)."""
py = python or os.environ.get("PYTHON") or "python3"
c.run(f"{_q(py)} {_q(script)} --help", pty=True)
使用方法¶
準備¶
msg/ 以下に.msgファイルを置く．
コマンド¶
$ invoke msg2json
$ invoke run
