Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 17 additions & 7 deletions src/investo/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -315,23 +315,29 @@ def _cmd_analyze(args: argparse.Namespace) -> int:
if args.json:
print(json.dumps(report.model_dump(), indent=2, default=str))

want_open = getattr(args, "open", False)

if want_html:
from .export import file_url, save_html
from .export import file_url, open_file, save_html
out = save_html(report, args.html) # args.html is None when the flag is bare
print(f"Wrote HTML report to {out}")
print(f" Open: {file_url(out)}") # clickable — opens in the browser on click
print(f" Location: {file_url(out)}")
if want_open:
open_file(out)

if want_pdf:
from .export import PdfExportError, file_url, save_pdf
from .export import PdfExportError, file_url, open_file, save_pdf
try:
out, engine, warnings = save_pdf(report, args.pdf)
for w in warnings:
print(f"warning: {w}", file=sys.stderr)
print(f"Wrote PDF report to {out} ({engine})")
print(f" Open: {file_url(out)}") # the pdf
print(f" Location: {file_url(out)}") # the pdf
sidecar = out.with_suffix(".html")
if sidecar.exists():
print(f" HTML: {file_url(sidecar)}") # the html written alongside it
print(f" HTML: {file_url(sidecar)}") # the html written alongside it
if want_open:
open_file(out)
except PdfExportError as exc:
# The .html sidecar is still on disk; a failed export is still a failed command.
print(f"error: PDF export failed.\n{exc}", file=sys.stderr)
Expand All @@ -341,10 +347,12 @@ def _cmd_analyze(args: argparse.Namespace) -> int:
# (--html / --pdf already produce a document) or opted out with --no-html. The announcement
# goes to stderr so `investo analyze X --json` stays pipeable on stdout.
if not (want_html or want_pdf or getattr(args, "no_html", False)):
from .export import file_url, save_html
from .export import file_url, open_file, save_html
out = save_html(report, None)
print(f"Wrote HTML report to {out}", file=sys.stderr)
print(f" Open: {file_url(out)}", file=sys.stderr)
print(f" Location: {file_url(out)}", file=sys.stderr)
if want_open:
open_file(out)

if not (args.json or want_html or want_pdf):
print(render_report(report))
Expand Down Expand Up @@ -401,6 +409,8 @@ def build_parser() -> argparse.ArgumentParser:
help="Write a PDF via headless Chrome/Edge (default name if FILE omitted)")
pa.add_argument("--no-html", action="store_true",
help="Suppress the HTML report that is otherwise written automatically")
pa.add_argument("--open", action="store_true",
help="Open the written report in your default app (browser / PDF viewer)")
_add_market(pa)
pa.set_defaults(func=_cmd_analyze)

Expand Down
83 changes: 80 additions & 3 deletions src/investo/export.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,17 @@

from __future__ import annotations

import http.server
import logging
import os
import shutil
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
from typing import TYPE_CHECKING
from urllib.parse import quote

from .config import CONFIG

Expand All @@ -36,13 +39,87 @@ def file_url(path: str | os.PathLike) -> str:
"""A clickable, absolute ``file://`` URL for a local path, percent-escaped.

``resolve().as_uri()`` is the same construction the headless-Chrome path uses to load the
report, so clicking this link opens exactly the file that was written (a space becomes ``%20``,
Windows drive letters get ``file:///C:/…``). Callers surface it so a reader can open the report
without hunting for the path.
report (a space becomes ``%20``, Windows drive letters get ``file:///C:/…``). Surfaced as the
report's on-disk *location*. Note a chat/webview will not *open* a ``file://`` link on click
(it is sandboxed); :func:`preview_url` returns an ``http://`` link that does.
"""
return Path(path).resolve().as_uri()


# --------------------------------------------------------------------------------------
# Local preview server: an http:// link the chat/webview will actually open on click
# --------------------------------------------------------------------------------------
# A file:// link is blocked from opening in a chat/webview; an http(s):// one is not. So each
# report directory gets a tiny loopback static server, and callers hand back an
# http://127.0.0.1:<port>/<file> URL that opens the *rendered* report in the browser on click.
_preview_lock = threading.Lock()
_preview_ports: dict[str, int] = {} # resolved directory -> its server port (one per directory)


def _start_preview_server(directory: Path) -> int:
"""Start a loopback static server rooted at ``directory`` on a daemon thread; return its port."""

class _Handler(http.server.SimpleHTTPRequestHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, directory=str(directory), **kwargs)

def log_message(self, *args): # keep the server quiet
pass

httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Handler) # port 0 -> OS assigns
threading.Thread(target=httpd.serve_forever, name="investo-preview", daemon=True).start()
return httpd.server_address[1]


def preview_url(path: str | os.PathLike) -> str | None:
"""A clickable ``http://127.0.0.1:<port>/<file>`` URL that opens the rendered report on click.

Serves the report's own directory over a loopback-only static server (started once per
directory, on a daemon thread). Unlike :func:`file_url`, this link opens from a chat/webview.
Best-effort: returns ``None`` (rather than raising) if a server can't be bound.
"""
try:
p = Path(path).resolve()
directory = str(p.parent)
with _preview_lock:
port = _preview_ports.get(directory)
if port is None:
port = _start_preview_server(p.parent)
_preview_ports[directory] = port
return f"http://127.0.0.1:{port}/{quote(p.name)}"
except OSError as exc: # binding failed / sandboxed network — degrade to no link
_log.warning("preview server unavailable: %s", exc)
return None


def _open_disabled() -> bool:
"""Never launch an app under tests, CI, or when the user opted out."""
return bool(os.environ.get("INVESTO_NO_OPEN") or os.environ.get("CI")
or os.environ.get("PYTEST_CURRENT_TEST"))


def open_file(path: str | os.PathLike) -> bool:
"""Open ``path`` in the OS default app (best-effort; never raises). Returns whether it tried.

Used by the CLI ``--open`` flag — a short-lived CLI can't host a persistent preview server, so
it opens the file directly instead.
"""
if _open_disabled():
return False
target = os.fspath(path)
try:
if sys.platform == "win32":
os.startfile(target) # noqa: S606 — opening our own generated report
elif sys.platform == "darwin":
subprocess.run(["open", target], check=False)
else:
subprocess.run(["xdg-open", target], check=False)
return True
except OSError as exc:
_log.warning("could not open %s: %s", target, exc)
return False


class PdfExportError(RuntimeError):
"""Raised when no PDF backend could produce a file. The message names the remedies."""

Expand Down
13 changes: 8 additions & 5 deletions src/investo/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -549,23 +549,25 @@ class AiSignals(_Base):
# Analysis tools: export, technicals, DCF sensitivity, multi-company compare
# --------------------------------------------------------------------------------------
class ExportedFile(_Base):
"""One written artifact and how to open it."""
"""One written artifact: where it is, and a link that opens it."""

path: str
file_url: str # clickable file:// URL — opens the file in the default app on click
file_url: str # file:// URL — the on-disk location (a chat/webview won't open this on click)
open_url: str | None = None # http://127.0.0.1 link that opens the rendered report on click
format: Literal["pdf", "html"]
bytes: int = 0


class ExportResult(_Base):
path: str
file_url: str | None = None # clickable file:// URL of `path`
file_url: str | None = None # file:// URL of `path` (on-disk location)
open_url: str | None = None # http://127.0.0.1 link that opens the rendered report on click
format: Literal["pdf", "html"]
bytes: int = 0
engine: str | None = None # e.g. "chrome.exe (headless)" | "playwright-chromium"
warnings: list[str] = Field(default_factory=list)
# Every artifact this call wrote, primary first — a PDF export also lists its HTML sidecar,
# so a caller has a clickable location for the html and the pdf.
# so a caller has a location and an open link for the html and the pdf.
files: list[ExportedFile] = Field(default_factory=list)


Expand Down Expand Up @@ -696,7 +698,8 @@ class AnalysisReport(_Base):
# Auto-export metadata: set when analyze_company writes an HTML report so a client can open
# it without a second tool call. Optional — the report is complete without them.
html_report_path: str | None = None
html_report_url: str | None = None # clickable file:// URL of html_report_path
html_report_url: str | None = None # file:// URL of html_report_path (on-disk location)
html_report_open_url: str | None = None # http://127.0.0.1 link that opens it on click
generated_at: str | None = None # ISO-8601 UTC timestamp of when the report was written
investo_version: str | None = None
html_bytes: int | None = None
Expand Down
18 changes: 10 additions & 8 deletions src/investo/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -137,14 +137,15 @@ def _attach_html_report(report: AnalysisReport) -> None:
from datetime import datetime, timezone

from . import __version__
from .export import default_filename, file_url, save_html
from .export import default_filename, file_url, preview_url, save_html

report.generated_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
report.investo_version = __version__
try:
out = save_html(report, _safe_export_path(default_filename(report, "html"), "html"))
report.html_report_path = str(out)
report.html_report_url = file_url(out) # clickable — opens the report on click
report.html_report_url = file_url(out) # file:// location
report.html_report_open_url = preview_url(out) # http link that opens it on click
report.html_bytes = out.stat().st_size
except Exception as exc: # noqa: BLE001 — the convenience export must never sink the analysis
_log.warning("automatic HTML export failed: %s", exc)
Expand Down Expand Up @@ -413,17 +414,18 @@ def export_report(
a caller cannot write outside it.
"""
from .analysis.report import analyze
from .export import PdfExportError, file_url, html_to_pdf, save_html
from .export import PdfExportError, file_url, html_to_pdf, preview_url, save_html

def _exported(p, fmt: Literal["pdf", "html"]) -> ExportedFile:
return ExportedFile(path=str(p), file_url=file_url(p), format=fmt, bytes=p.stat().st_size)
return ExportedFile(path=str(p), file_url=file_url(p), open_url=preview_url(p),
format=fmt, bytes=p.stat().st_size)

out = _safe_export_path(path, format)
report = analyze(query, market)

if format == "html":
f = _exported(save_html(report, out), "html")
return ExportResult(path=f.path, file_url=f.file_url, format="html",
return ExportResult(path=f.path, file_url=f.file_url, open_url=f.open_url, format="html",
bytes=f.bytes, engine="renderer", files=[f])

from .render import render_html
Expand All @@ -434,10 +436,10 @@ def _exported(p, fmt: Literal["pdf", "html"]) -> ExportedFile:
engine, warnings = html_to_pdf(html, out)
except PdfExportError as exc:
raise ValueError(f"{exc}") from exc
# Return a clickable location for the pdf and its html sidecar (primary first).
# Return the location and an open link for the pdf and its html sidecar (primary first).
pdf, html_side = _exported(out, "pdf"), _exported(sidecar, "html")
return ExportResult(path=pdf.path, file_url=pdf.file_url, format="pdf", bytes=pdf.bytes,
engine=engine, warnings=warnings, files=[pdf, html_side])
return ExportResult(path=pdf.path, file_url=pdf.file_url, open_url=pdf.open_url, format="pdf",
bytes=pdf.bytes, engine=engine, warnings=warnings, files=[pdf, html_side])


@mcp.tool(title="Full analysis", annotations=_WRITE)
Expand Down
8 changes: 8 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,14 @@ def test_no_html_flag_suppresses_the_automatic_report(tmp_path, monkeypatch, cap
assert not list(tmp_path.glob("*.html")), "--no-html must write no HTML file"


def test_open_flag_is_accepted_and_still_writes_the_report(tmp_path, monkeypatch, capsys):
# --open opens the report in the default app; under pytest open_file is a guarded no-op, so the
# command succeeds without launching anything and the file is still written.
monkeypatch.chdir(tmp_path)
assert _run(["analyze", "KPIT", "--open"]) == 0
assert list(tmp_path.glob("investo-KPITTECH.NS-*.html"))


def test_explicit_pdf_does_not_also_auto_write_html(tmp_path, monkeypatch, capsys):
# --pdf already produces a document (and its own .html sidecar); don't also drop a second
# auto-HTML in the cwd.
Expand Down
27 changes: 25 additions & 2 deletions tests/test_export.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,8 @@ def test_windows_style_path_becomes_a_percent_escaped_file_uri():
assert "%20" in uri # the space is escaped, not left raw


def test_file_url_is_an_absolute_clickable_uri(tmp_path):
# The clickable location returned to the user: absolute, percent-escaped, opens on click.
def test_file_url_is_an_absolute_location_uri(tmp_path):
# The on-disk location returned to the user: absolute, percent-escaped file:// URI.
target = tmp_path / "a b" / "investo-report.html"
target.parent.mkdir(parents=True)
target.write_text("<html></html>", encoding="utf-8")
Expand All @@ -84,6 +84,29 @@ def test_file_url_is_an_absolute_clickable_uri(tmp_path):
assert url.endswith("investo-report.html")


def test_preview_url_serves_the_report_over_loopback_http(tmp_path):
# A file:// link won't open from a chat/webview; this http://127.0.0.1 one does. Prove it by
# fetching the report back over the loopback preview server.
import urllib.request

target = tmp_path / "investo-report.html"
target.write_text("<html>hello sigachi</html>", encoding="utf-8")
url = export.preview_url(target)
assert url and url.startswith("http://127.0.0.1:")
assert url.endswith("investo-report.html")
with urllib.request.urlopen(url, timeout=5) as resp: # noqa: S310 — loopback, our own server
assert resp.status == 200
assert b"hello sigachi" in resp.read()


def test_open_file_is_a_guarded_no_op_under_tests(tmp_path):
# PYTEST_CURRENT_TEST is set during tests, so open_file must not launch anything (and returns
# False rather than raising) — no browser windows during the suite.
target = tmp_path / "r.html"
target.write_text("x", encoding="utf-8")
assert export.open_file(target) is False


# --------------------------------------------------------------------------------------
# Headless Chrome path — stubbed subprocess, never a real launch
# --------------------------------------------------------------------------------------
Expand Down
4 changes: 3 additions & 1 deletion tests/test_server_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,9 +110,11 @@ def test_attach_html_report_writes_into_the_sandbox_and_records_metadata(tmp_pat
assert written.read_text(encoding="utf-8").startswith("<!doctype html>")
assert report.html_bytes and report.html_bytes > 0
assert report.investo_version and report.generated_at
# A clickable location so a client can open the report without a second call.
# The on-disk location, plus an http link that opens the rendered report on click.
assert report.html_report_url and report.html_report_url.startswith("file:///")
assert report.html_report_url == written.resolve().as_uri()
assert (report.html_report_open_url
and report.html_report_open_url.startswith("http://127.0.0.1:"))


# --------------------------------------------------------------------------------------
Expand Down
Loading