(user_id: str, export_format: str = "markdown")
| 1281 | root.mkdir(parents=True, exist_ok=True) |
| 1282 | return root |
| 1283 | |
| 1284 | |
| 1285 | def _export_sources() -> List[Tuple[str, Path]]: |
| 1286 | return [ |
| 1287 | ("papers", Path(_configured_path("PAPERFLOW_PDF_DIR", "data/exports")).resolve()), |
| 1288 | ("reading_reports", Path(_configured_path("PAPERFLOW_READING_REPORTS_DIR", "data/exports")).resolve()), |
| 1289 | ("wiki", Path(_configured_path("PAPERFLOW_WIKI_DIR", "data/wiki")).resolve()), |
| 1290 | ] |
| 1291 | |
| 1292 | |
| 1293 | def export_data(user_id: str, export_format: str = "markdown") -> Dict[str, Any]: |
| 1294 | normalized = str(export_format or "markdown").strip().lower() |
| 1295 | timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| 1296 | root = _export_root() |
| 1297 | |
| 1298 | if normalized in {"markdown", "md"}: |
| 1299 | output = root / f"paperflow_{user_id or 'all'}_{timestamp}.md" |
| 1300 | reports = list_reports(user_id=user_id, days=3650, limit=200).get("reports") or [] |
| 1301 | wiki_nodes = wiki_db.list_nodes(user_id, limit=200) if user_id else [] |
| 1302 | filtered_wiki_nodes = _filter_wiki_nodes_for_configured_dir(wiki_nodes) |
| 1303 | if filtered_wiki_nodes is not None: |
| 1304 | wiki_nodes = filtered_wiki_nodes |
| 1305 | lines = [ |
| 1306 | f"# PaperFlow Export - {user_id or 'all'}", |
| 1307 | "", |
| 1308 | f"- Exported at: {datetime.now().isoformat(sep=' ', timespec='seconds')}", |
| 1309 | f"- Reports: {len(reports)}", |
| 1310 | f"- Wiki nodes: {len(wiki_nodes)}", |
| 1311 | "", |
| 1312 | "## Reading Reports", |
| 1313 | "", |
| 1314 | ] |
| 1315 | for report in reports: |
| 1316 | lines.extend( |
| 1317 | [ |
| 1318 | f"### {report.get('title') or 'Untitled Report'}", |
| 1319 | "", |
| 1320 | f"- Path: {report.get('report_path') or ''}", |
| 1321 | f"- Saved: {report.get('saved_at') or ''}", |
| 1322 | "", |
| 1323 | str(report.get("snippet") or "").strip(), |
| 1324 | "", |
| 1325 | ] |
| 1326 | ) |
| 1327 | lines.extend(["## Wiki Nodes", ""]) |
| 1328 | for node in wiki_nodes: |
| 1329 | lines.extend( |
| 1330 | [ |
| 1331 | f"### {node.get('title') or node.get('node_id')}", |
| 1332 | "", |
| 1333 | f"- Type: {node.get('node_type') or ''}", |
| 1334 | f"- Path: {node.get('file_path') or ''}", |
| 1335 | "", |
| 1336 | str(node.get("body") or "")[:1200], |
| 1337 | "", |
| 1338 | ] |
| 1339 | ) |
| 1340 | output.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8") |
nothing calls this directly
no test coverage detected