diff --git a/adapters/__init__.py b/adapters/__init__.py deleted file mode 100644 index c84fa5b..0000000 --- a/adapters/__init__.py +++ /dev/null @@ -1,12 +0,0 @@ -"""adapters/ — Adapter riêng cho Qt (clock, thread, timer). - -Kế hoạch gốc đặt tên thư mục này là ``platform/``. Không dùng được: chạy -bất kỳ script nào từ thư mục gốc repo (``python tools/...``, -``python scripts/...``) thì ``platform/`` **che khuất module ``platform`` -của thư viện chuẩn**, và ``import keyring`` chết ngay với -``AttributeError: module 'platform' has no attribute 'system'``. -Repo có 26 script chạy đúng kiểu đó. - -Đổi tên là cách duy nhất chắc chắn — không thể bắt mọi người nhớ "đừng bao -giờ chạy python từ thư mục gốc". -""" diff --git a/adapters/qt/__init__.py b/adapters/qt/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/application/settings/__init__.py b/application/settings/__init__.py deleted file mode 100644 index c759b25..0000000 --- a/application/settings/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Application settings package: Settings application service.""" diff --git a/infrastructure/platform/__init__.py b/infrastructure/platform/__init__.py deleted file mode 100644 index 9115969..0000000 --- a/infrastructure/platform/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Infrastructure platform adapters package.""" diff --git a/infrastructure/platform/qt/__init__.py b/infrastructure/platform/qt/__init__.py deleted file mode 100644 index ca02309..0000000 --- a/infrastructure/platform/qt/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Infrastructure Qt platform adapters: QtSchedulerClock.""" diff --git a/presentation/folder/ai_edit_runner.py b/presentation/folder/ai_edit_runner.py deleted file mode 100644 index 088d6a0..0000000 --- a/presentation/folder/ai_edit_runner.py +++ /dev/null @@ -1,325 +0,0 @@ -"""Một lượt AI sửa file, từ lúc gửi tới lúc ghi ra đĩa — R08-T12. - -``_ai_run_edit`` dài (82 dòng) vì nó là cả một lượt: dựng ngữ cảnh từ -file đang mở, gọi provider, nhận nội dung phát dần, tách phần mã khỏi -phần giải thích, rồi dựng bản xem trước. - -Không bao giờ ghi đè thẳng: kết quả hiện ra để người dùng xem, và chỉ -``_ai_apply`` mới chạm vào file. - -Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``. -""" -from __future__ import annotations - -from .file_helpers import ( - _HTML_SUFFIXES, _PPTX_SUFFIXES, _parse_ai_output, _pptx_available, -) - -import os -from pathlib import Path -from typing import Optional -from PySide6.QtCore import Qt -from ...core.worker import AgentWorker -from ...i18n import tr -from ...theme import current_palette - - -class AiEditRunnerMixin: - """Trộn vào FolderTab.""" - - def _ai_send(self) -> None: - if not self._root or not os.path.isdir(self._root): - self.ai_chat.add_error(tr("folder.ai_no_file")) - return - instruction = self.ai_input.text().strip() - if not instruction: - return - self.ai_input.clear() - self.ai_chat.add_user(instruction) - # QUEUE: while a run is active OR a proposal is awaiting Apply/Discard, - # hold the new instruction and run it when the pipeline goes idle. Lets - # the user line up several edits without waiting for each to finish. - if self._ai_worker is not None or self._ai_pending is not None: - self._ai_queue.append(instruction) - self.ai_chat.add_status(tr("folder.ai_queued", n=len(self._ai_queue))) - self._update_queue_status() - return - self._ai_start(instruction) - - def _ai_start(self, instruction: str) -> None: - """Begin processing one instruction (plan → edit). Assumes the pipeline - is idle (the queue calls this when the previous run finishes).""" - # If a text/code/HTML file is open (even in Preview), switch it into the - # editor so AI can edit it. If nothing editable is open, that's fine — - # the request may be to CREATE a new file (the model names it via FILE:). - editable = self.stack.currentWidget() is self.editor - if not editable: - editable = self._ensure_editor_for_ai() - self._maybe_suggest_image_model(instruction) - # Auto Model Routing (may switch to the best coding model for this run). - self._ai_apply_routing(instruction) - has_file = editable and bool(self._current_file) - self._ai_running_file = Path(self._current_file).name if has_file else tr("folder.ai_new_file") - self._ai_set_busy(True) - # Announce start on the status bar so it's visible even from another tab — - # the edit keeps running in the background until it finishes. - self.status_message.emit(tr("folder.ai_running", name=self._ai_running_file)) - # Two phases so the PLAN is shown INLINE *before* the edit runs. - self._ai_ctx = { - "filename": Path(self._current_file).name if has_file else "", - "content": self.editor.toPlainText() if has_file else "", - "convo": self._cowork_context(), - "instruction": instruction, - "provider": self._ai_provider(), - "plan": "", - } - # Reset the token/cost tally for THIS prompt (plan + edit calls sum into it). - self._ai_prompt_usage = {"in": 0, "out": 0, "cache": 0, "cost": 0.0} - self._ai_run_plan() - - def _ai_maybe_dequeue(self) -> None: - """When the pipeline is fully idle, start the next queued instruction.""" - if self._ai_worker is not None or self._ai_pending is not None: - return - if not self._ai_queue: - return - nxt = self._ai_queue.pop(0) - self._update_queue_status() - self._ai_start(nxt) - - def _ai_add_usage(self, usage) -> None: - """Add one model call's usage (plan or edit) to THIS prompt's tally.""" - if not isinstance(usage, dict): - return - tot = getattr(self, "_ai_prompt_usage", None) - if tot is None: - tot = self._ai_prompt_usage = {"in": 0, "out": 0, "cache": 0, "cost": 0.0} - tot["in"] += int(usage.get("in", 0) or 0) - tot["out"] += int(usage.get("out", 0) or 0) - tot["cache"] += int(usage.get("cache", 0) or 0) - tot["cost"] += float(usage.get("cost_usd", 0.0) or 0.0) - - def _ai_show_usage(self, bubble) -> None: - """Footer under the AI-edit reply: ↓in ↑out ▤ctx $cost for the whole - prompt (plan + edit), priced in the display currency — same as Cowork.""" - tot = getattr(self, "_ai_prompt_usage", None) - if bubble is None or not tot or not (tot["in"] or tot["out"]): - return - from ...core import model_pricing as mp, usage_tracker as ut - pricing = {**ut.DEFAULT_PRICING, **(self.ctx.config.data.get("usage") or {})} - line = (f"↓{mp.format_tokens(int(tot['in']))} ↑{mp.format_tokens(int(tot['out']))} " - f"▤{mp.format_tokens(int(tot['in'] + tot['out'] + tot['cache']))} " - f"{ut.format_cost(tot['cost'], pricing)}") - try: - bubble.add_usage(line) - except Exception: # noqa: BLE001 - a usage footer must never break the edit - pass - - def _ai_run_plan(self) -> None: - c = self._ai_ctx - plan_bubble = self.ai_chat.add_plan(tr("folder.ai_planning")) - self.ai_chat.scroll_to_bottom() - - def job(worker): - from ...core import usage_tracker as ut - from ...core.co4e_runner import _usage_delta - provider = c["provider"] - messages = [{"role": "system", "content": - "You are an AI file editor. Give a SHORT numbered plan (2-4 steps) for " - "the requested change. Plan ONLY — do NOT output any code."}] - if c["convo"]: - messages.append({"role": "system", - "content": "Context from the user's Cowork conversation:\n" + c["convo"]}) - messages.append({"role": "user", "content": - f"File: {c['filename']}\n\nCurrent content:\n```\n{c['content']}\n```\n\n" - f"Request: {c['instruction']}"}) - ut.set_context("folder", c.get("filename") or "AI edit") # Dashboard + usage - ut.begin_accumulation(); base = ut.accumulated() - try: - r = provider.chat(messages, tools=None, cancel=worker.is_cancelled) - txt = r.get("content", "") if isinstance(r, dict) else str(r) - usage = _usage_delta(base, self.ctx.config) - finally: - ut.end_accumulation() - return {"plan": provider.strip_think(txt) or "", "usage": usage} - - worker = AgentWorker(job) - worker.finished_ok.connect(lambda res, b=plan_bubble: self._ai_plan_done(res, b)) - worker.failed.connect(lambda err, b=plan_bubble: self._ai_failed(err, b)) - self._ai_worker = worker - worker.start() - - def _ai_plan_done(self, result, plan_bubble) -> None: - self._ai_add_usage((result or {}).get("usage")) # plan-step tokens - plan = ((result or {}).get("plan") or "").strip() - self._ai_ctx["plan"] = plan - plan_bubble.set_plain(plan or tr("folder.ai_empty")) - self.ai_chat.scroll_to_bottom() - self._ai_run_edit() # now execute the plan - - def _ai_run_edit(self) -> None: - c = self._ai_ctx - bubble = self.ai_chat.add_assistant(tr("folder.ai_edit")) - self.ai_chat.scroll_to_bottom() - - pptx_note = ("\nThis is a PPTX shown as marker blocks '### Slide N / Box M', where N is the " - "1-based SLIDE NUMBER and M the box on that slide. When the user refers to a " - "slide by number (e.g. 'edit slide 3'), change ONLY the blocks under '### Slide " - "3' and leave every other slide's block exactly as-is. Each block has fields " - "type/pos/size/font/text (and image for pictures). To change TEXT COLOUR or " - "FONT edit the `font:` line — e.g. `font: name=Arial size=28 bold=1 " - "color=FF0000`. Keep all block markers and structure.") if self._edit_kind == "pptx" else "" - - # When creating a NEW deck (request mentions slides/pptx and we're not - # already editing one), tell the model the marker format to emit so we can - # build a real .pptx from it. - _pptx_words = ("pptx", "powerpoint", "slide", "presentation", "deck", - "スライド", "プレゼン", "trình chiếu", "trinh chieu", "bài thuyết trình") - wants_new_pptx = (self._edit_kind != "pptx" - and any(w in c["instruction"].lower() for w in _pptx_words)) - new_pptx_note = ("\nTo CREATE a PowerPoint, name it `FILE: .pptx` and output the slides " - "as marker blocks — one block per shape:\n" - "### Slide 1 / Box 1\ntype: text\npos: 0.5, 0.4\nsize: 9.0, 1.2\n" - "font: name=Calibri size=32 bold=1 color=1F3864\ntext:\nTitle here\n\n" - "### Slide 1 / Box 2\ntype: text\npos: 0.5, 1.8\nsize: 9.0, 4.5\n" - "text:\nBullet one\nBullet two\n\n" - "Increment the Slide number for each new slide; pos/size are in inches; " - "font color is RRGGBB hex.") if wants_new_pptx else "" - - imggen_note = "" - try: - from ...core import image_gen - if image_gen.is_configured(self.ctx.config): - imggen_note = ("\nYou can also GENERATE an illustration image: add a line " - "`IMAGE_GEN: => `. Use a " - "generated image e.g. as a new picture, or (for pptx) set a picture " - "box's `image:` field to that same path to insert it.") - except Exception: # noqa: BLE001 - pass - - def job(worker): - provider = c["provider"] - open_note = (f"the currently-open file '{c['filename']}'" if c["filename"] - else "no file is open") - messages = [{"role": "system", "content": - "You are an AI file editor inside an app. Following the plan, output the " - "COMPLETE file content in ONE fenced code block (```), and nothing after " - "it. Preserve everything you were not asked to change.\n" - "If the request is to CREATE A NEW file (or a different file than the one " - "open), put a line `FILE: ` (relative to the " - "current folder) immediately before the code block. Omit FILE to edit the " - f"open file. Right now, {open_note}." + pptx_note + new_pptx_note + imggen_note}] - if c["convo"]: - messages.append({"role": "system", - "content": "Context from the user's Cowork conversation:\n" + c["convo"]}) - if c["plan"]: - messages.append({"role": "system", "content": "Plan to follow:\n" + c["plan"]}) - cur = (f"File: {c['filename']}\n\nCurrent content:\n```\n{c['content']}\n```\n\n" - if c["filename"] else "No file is currently open.\n\n") - messages.append({"role": "user", "content": cur + f"Request: {c['instruction']}"}) - - def on_text(piece: str) -> None: - worker.emit_event({"type": "text", "delta": piece}) - - from ...core import usage_tracker as ut - from ...core.co4e_runner import _usage_delta - ut.set_context("folder", c.get("filename") or "AI edit") # Dashboard + usage - ut.begin_accumulation(); base = ut.accumulated() - try: - r = provider.chat(messages, tools=None, on_text=on_text, cancel=worker.is_cancelled) - txt = r.get("content", "") if isinstance(r, dict) else str(r) - usage = _usage_delta(base, self.ctx.config) - finally: - ut.end_accumulation() - return {"text": provider.strip_think(txt) or "", "usage": usage} - - worker = AgentWorker(job) - worker.event.connect(lambda ev, b=bubble: self._ai_stream(ev, b)) - worker.finished_ok.connect(lambda res, b=bubble: self._ai_done(res, b)) - worker.failed.connect(lambda err, b=bubble: self._ai_failed(err, b)) - self._ai_worker = worker - worker.start() - - def _ai_stream(self, ev, bubble) -> None: - if isinstance(ev, dict) and ev.get("type") == "text": - bubble.append_delta(ev.get("delta", "")) - self.ai_chat.scroll_to_bottom() - - def _ai_done(self, result, bubble) -> None: - self._ai_worker = None - self._ai_set_busy(False) - self._ai_add_usage((result or {}).get("usage")) # edit-step tokens - self._ai_show_usage(bubble) # footer: prompt total (plan+edit) - text = ((result or {}).get("text") or "").strip() - target, new_content, summary, image_gens = _parse_ai_output(text) - if new_content is None and not image_gens: - bubble.set_markdown(text or tr("folder.ai_empty")) - self.ai_chat.scroll_to_bottom() - self._ai_flag_done() - return - # Decide edit-current vs create-new. A FILE: naming a path different from - # the open file (or when nothing is open) → CREATE a new file. - create = bool(target) and (not self._current_file - or Path(target).name != Path(self._current_file).name) - # PROPOSE the change — nothing is written until the user clicks Apply. - self._ai_pending = {"content": new_content, - "target": target if create else None, - "image_gens": image_gens} - hint = tr("folder.ai_review_hint") - bubble.set_markdown((summary + "\n\n" if summary else "") + "_" + hint + "_") - if new_content is not None: - import difflib - old = "" if create else self.editor.toPlainText() - diff = "".join(difflib.unified_diff( - old.splitlines(keepends=True), new_content.splitlines(keepends=True), - fromfile=("(new file)" if create else "current"), - tofile=(target if create else "proposed"))) or "(no textual difference)" - title = tr("folder.ai_proposed_new", name=target) if create else tr("folder.ai_proposed") - self.ai_chat.add_diff(title, diff) - if image_gens: - listing = "\n".join(f"• {p} ({prompt[:60]})" for prompt, p in image_gens) - self.ai_chat.add_status(tr("folder.ai_image_plan") + "\n" + listing) - self._ai_confirm_row.setVisible(True) - self.ai_chat.scroll_to_bottom() - name = target if create else getattr(self, "_ai_running_file", "") - self.status_message.emit(tr("folder.ai_proposed_status", name=name)) - self._ai_status.setText("● " + hint) - self._ai_status.setStyleSheet(f"color:{current_palette().warning};") - - def _ai_apply(self) -> None: - """Confirmed by the user. If the edit GENERATES images, ask the image - gate then generate them (off-thread) before finalising the file edit.""" - if not self._ai_pending: - return - p = self._ai_pending - self._ai_pending = None - self._ai_confirm_row.setVisible(False) - if p.get("image_gens"): - from PySide6.QtWidgets import QMessageBox - if QMessageBox.question(self, tr("folder.ai_image_confirm_title"), - tr("folder.ai_image_confirm_gen")) != QMessageBox.Yes: - self.status_message.emit(tr("folder.ai_image_declined")) - return - self._ai_generate_then_finalize(p) - return - self._ai_finalize_apply(p) - - - - - - def _ai_discard(self) -> None: - self._ai_pending = None - self._ai_confirm_row.setVisible(False) - self.ai_chat.add_status(tr("folder.ai_discarded")) - self.ai_chat.scroll_to_bottom() - self._ai_status.setText("") - self._ai_maybe_dequeue() # discarding resolves the gate → run the next queued edit - - - def _ai_failed(self, err, bubble) -> None: - self._ai_worker = None - bubble.set_markdown(tr("folder.ai_error", err=err)) - self._ai_set_busy(False) - self.status_message.emit(tr("folder.ai_error", err=err)) - self._ai_flag_done() diff --git a/presentation/folder/ai_output_writer.py b/presentation/folder/ai_output_writer.py deleted file mode 100644 index 6581853..0000000 --- a/presentation/folder/ai_output_writer.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Ghi kết quả AI ra đĩa — R08-T12. - -Tách khỏi ``ai_edit_runner.py`` vì đây là phần DUY NHẤT thật sự chạm vào -file của người dùng. Mọi thứ trước nó chỉ dựng bản xem trước. - -Gồm cả nhánh sinh ảnh: lượt nào có ảnh thì phải chờ ảnh xong mới ghi, vì -nội dung có thể tham chiếu tới đường dẫn ảnh vừa tạo. -""" -from __future__ import annotations - -from .file_helpers import ( - _HTML_SUFFIXES, _PPTX_SUFFIXES, _parse_ai_output, _pptx_available, -) -import os -from pathlib import Path -from typing import Optional -from PySide6.QtCore import Qt -from ...core.worker import AgentWorker -from ...i18n import tr -from ...theme import current_palette - - -class AiOutputWriterMixin: - """Trộn vào FolderTab.""" - - def _ai_generate_then_finalize(self, p: dict) -> None: - imgs = p.get("image_gens") or [] - root = os.path.normpath(self._root) - img_model, img_base, img_key = self._ai_image_model() # may target another provider - self._ai_set_busy(True) - self.status_message.emit(tr("folder.ai_generating")) - - def job(worker): - from ...core import image_gen - results = [] - for prompt, rel in imgs: - dest = rel if os.path.isabs(rel) else os.path.join(root, rel) - dest = os.path.normpath(dest) - if os.path.commonpath([dest, root]) != root: - results.append((rel, False, "path escapes the folder")) - continue - try: - os.makedirs(os.path.dirname(dest) or root, exist_ok=True) - except OSError as exc: - results.append((rel, False, str(exc))) - continue - ok, msg = image_gen.generate_image(self.ctx.config, prompt, dest, - model=img_model, base_url=img_base, api_key=img_key) - results.append((dest, ok, msg)) - return {"results": results} - - worker = AgentWorker(job) - worker.finished_ok.connect(lambda res, pp=p: self._ai_images_done(res, pp)) - worker.failed.connect(lambda err, pp=p: self._ai_images_done({"results": [], "err": err}, pp)) - self._ai_worker = worker - worker.start() - - def _ai_images_done(self, res: dict, p: dict) -> None: - self._ai_worker = None - self._ai_set_busy(False) - created = [] - for dest, ok, msg in res.get("results", []): - if ok: - created.append(dest) - self.ai_chat.add_success("✓ " + tr("folder.ai_image_created", name=Path(dest).name)) - else: - self.ai_chat.add_error(tr("folder.ai_image_failed", err=msg)) - # Now apply any text/file edit (pptx image: fields now point at real files). - self._ai_finalize_apply(p, images_done=True) - # If it was only image generation, open the first new image. - if p.get("content") is None and not p.get("target") and created: - self.open_file(created[0], reset=False) - - def _ai_finalize_apply(self, p: dict, images_done: bool = False) -> None: - content = p.get("content") - target = p.get("target") - if content is None: - self.ai_chat.scroll_to_bottom() - self._ai_flag_done() - self.status_message.emit(tr("folder.ai_done", name=getattr(self, "_ai_running_file", ""))) - return - if target: - dest = self._create_new_file(target, content) - if dest is None: - return - self.ai_chat.add_success("✓ " + tr("folder.ai_created", name=Path(dest).name)) - self.status_message.emit(tr("folder.ai_created", name=Path(dest).name)) - else: - self.editor.setPlainText(content) # live update in the editor/preview - self._ai_write_out(content, skip_image_confirm=images_done) - self.ai_chat.add_success("✓ " + tr("folder.ai_applied")) - self.status_message.emit(tr("folder.ai_done", name=getattr(self, "_ai_running_file", ""))) - self.ai_chat.scroll_to_bottom() - self._ai_flag_done() - - def _create_new_file(self, target: str, content: str) -> Optional[str]: - """Create ``target`` (relative to the folder root) with ``content`` and - open it — like Cowork's save_file. Refuses paths escaping the root.""" - root = os.path.normpath(self._root) - dest = target if os.path.isabs(target) else os.path.join(root, target) - dest = os.path.normpath(dest) - if os.path.commonpath([dest, root]) != root: - self.status_message.emit(tr("folder.ai_error", err="path escapes the folder")) - return None - try: - os.makedirs(os.path.dirname(dest) or root, exist_ok=True) - if Path(dest).suffix.lower() in _PPTX_SUFFIXES and _pptx_available(): - # A .pptx is a binary package — build a real deck from the marker - # text (writing text straight to .pptx would corrupt it). - from ...core import pptx_edit - pptx_edit.create_pptx_from_text(dest, content) - else: - Path(dest).write_text(content, encoding="utf-8") - except Exception as exc: # noqa: BLE001 - OS error or pptx build failure - self.status_message.emit(tr("folder.save_error", err=str(exc))) - return None - self.open_file(dest, reset=False) # show the new file; keep this AI chat - return dest - - def _ai_write_out(self, content: str, skip_image_confirm: bool = False) -> None: - """Persist the confirmed content to disk AND refresh the preview. - pptx text is written back into the deck (no PowerPoint window).""" - if not self._current_file: - return - try: - if self._edit_kind == "pptx": - if not self._write_pptx(content, skip_confirm=skip_image_confirm): - return - else: - Path(self._current_file).write_text(content, encoding="utf-8") - except Exception as exc: # noqa: BLE001 - self.status_message.emit(tr("folder.save_error", err=str(exc))) - return - # Refresh preview: HTML re-renders; pptx re-renders the slides; code stays - # in the (now-saved) editor. - suffix = Path(self._current_file).suffix.lower() - if suffix in _HTML_SUFFIXES: - self._show_html(self._current_file, mode_preview=True) - elif suffix in _PPTX_SUFFIXES: - self._show_pptx(self._current_file, mode_preview=True) diff --git a/presentation/folder/file_helpers.py b/presentation/folder/file_helpers.py deleted file mode 100644 index 8241830..0000000 --- a/presentation/folder/file_helpers.py +++ /dev/null @@ -1,114 +0,0 @@ -"""Hàm phụ trợ đọc và nhận dạng file — R08-T12. - -Thuần hàm, không widget. ``_is_probably_text`` là chỗ quyết định file -mở bằng ô soạn thảo hay báo là nhị phân — đoán sai thì người dùng thấy -một màn hình ký tự rác. -""" -from __future__ import annotations - -from ...ui.libreoffice_view import DOC_SUFFIXES # noqa: F401 — dùng lại ở cả gói - -# Nhận dạng loại file và các ngưỡng — gom về đây vì cả sáu file trong gói -# đều hỏi tới, để rải ra thì mỗi lần thêm một đuôi file phải sửa vài chỗ. -try: - from ...graph.graph_web import _HAS_WEB -except Exception: # pragma: no cover - _HAS_WEB = False - -try: - from PySide6.QtPdf import QPdfDocument # noqa: F401 - from PySide6.QtPdfWidgets import QPdfView # noqa: F401 - _HAS_PDF = True -except Exception: # pragma: no cover - QtPdf not bundled - _HAS_PDF = False - -_IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".ico"} -_HTML_SUFFIXES = {".html", ".htm"} -_PPTX_SUFFIXES = {".pptx"} # text-editable in-app via python-pptx (no PowerPoint) -_EXCEL_SUFFIXES = {".xlsx", ".xlsm"} # viewed as a real table via openpyxl (no LibreOffice) -_MAX_EDIT_BYTES = 2_000_000 # above this, view read-only / note only -_MAX_HIGHLIGHT_CHARS = 300_000 # skip colouring huge files (keeps typing snappy) - - -import os -from pathlib import Path -from PySide6.QtCore import Qt -from PySide6.QtGui import QColor, QFont, QTextCharFormat -from ...i18n import tr - - -def _fmt(color: str, *, italic: bool = False, bold: bool = False) -> QTextCharFormat: - f = QTextCharFormat() - f.setForeground(QColor(color)) - if italic: - f.setFontItalic(True) - if bold: - f.setFontWeight(QFont.Bold) - return f - - -def _pptx_available() -> bool: - """True when python-pptx is importable. If it's MISSING, auto-download & - install it (via deps.ensure_module) so pptx editing 'just works' — cached so - the (one-time) install is attempted only once.""" - global _PPTX_READY - if _PPTX_READY is None: - try: - from ...core.deps import ensure_module - _PPTX_READY = ensure_module("pptx", "python-pptx") is not None - except Exception: # noqa: BLE001 - _PPTX_READY = False - return _PPTX_READY - - -def _split_code_block(text: str): - """Split an AI reply into ``(file_content, summary)``. ``file_content`` is - the first fenced code block (the edited file); ``summary`` is any prose - before it. Returns ``(None, text)`` when there's no code block.""" - import re - m = re.search(r"```[^\n]*\n(.*?)```", text or "", re.DOTALL) - if not m: - return None, (text or "") - return m.group(1), (text[:m.start()].strip()) - - -def _parse_ai_output(text: str): - """Parse an AI edit reply into ``(target, content, summary, image_gens)``. - ``FILE: `` names a NEW file to create; ``IMAGE_GEN: => `` - lines request generated illustration images (relative paths).""" - import re - content, summary = _split_code_block(text) - target = None - m = re.search(r"(?mi)^\s*FILE:\s*(.+?)\s*$", text or "") - if m: - target = m.group(1).strip().strip("`\"'") - image_gens = [] - for gm in re.finditer(r"(?mi)^\s*IMAGE_GEN:\s*(.+?)\s*=>\s*(\S+)\s*$", text or ""): - image_gens.append((gm.group(1).strip(), gm.group(2).strip().strip("`\"'"))) - # Strip the directive lines out of the shown summary. - summary = re.sub(r"(?mi)^\s*(FILE|IMAGE_GEN):\s*.+?$", "", summary).strip() - return target, content, summary, image_gens - - -def _read_text(path: str) -> str: - try: - return Path(path).read_text(encoding="utf-8", errors="replace") - except OSError as exc: - return f"[could not read file: {exc}]" - - -def _is_probably_text(path: str) -> bool: - try: - with open(path, "rb") as f: - chunk = f.read(4096) - except OSError: - return False - if b"\x00" in chunk: - return False - try: - chunk.decode("utf-8") - return True - except UnicodeDecodeError: - # Latin-ish text still edits fine via errors="replace"; only reject on - # a hard binary signal (NUL above), so most source files pass. - return True diff --git a/presentation/folder/image_model_picker.py b/presentation/folder/image_model_picker.py deleted file mode 100644 index 095c42f..0000000 --- a/presentation/folder/image_model_picker.py +++ /dev/null @@ -1,115 +0,0 @@ -"""Chọn model sinh ảnh cho AI sửa file — R08-T12. - -Dò khắp mọi provider đã cấu hình xem cái nào sinh được ảnh, rồi gợi ý khi -câu người dùng gõ nghe như đang muốn tạo ảnh. Có thể gợi ý model của -provider KHÁC provider đang chọn — nên nó tách riêng: đây là chỗ duy nhất -trong màn Thư mục biết tới nhiều provider cùng lúc. -""" -from __future__ import annotations - -from .file_helpers import ( - DOC_SUFFIXES, _HTML_SUFFIXES, _IMAGE_SUFFIXES, _PPTX_SUFFIXES, _is_probably_text, _pptx_available, -) -import os -from pathlib import Path -from PySide6.QtCore import Qt, Signal -from PySide6.QtWidgets import QComboBox, QHBoxLayout, QLabel, QLineEdit, QPushButton, QVBoxLayout, QWidget -from ...core.worker import AgentWorker -from ...i18n import tr -from ...theme import current_palette -from ...ui.chat_view import ChatView -from ...ui.libreoffice_view import DOC_SUFFIXES - - -class ImageModelPickerMixin: - """Trộn vào FolderTab.""" - - def _scan_all_image_models(self, then_suggest: bool = False) -> None: - """Background: find image-capable models across EVERY configured provider - (not just the active one), so we can suggest one when an edit involves - images even if the active provider has none. Caches - ``self._all_image_models = [(provider_key, model)]``.""" - if self._img_scan_worker is not None: - if then_suggest: - self._pending_img_suggest = True - return - providers = dict(self.ctx.config.data.get("providers", {})) - # Only providers that actually have an endpoint/key configured. - candidates = [k for k, c in providers.items() - if (c.get("base_url") or c.get("api_key"))] - - def job(worker): - from ...core import image_gen - found = [] - for key in candidates: - try: - prov = self.ctx.build_provider_for(key) - models = list(getattr(prov, "list_models", lambda: [])() or []) - except Exception: # noqa: BLE001 - a broken provider must not block the scan - models = [] - for m in models: - if image_gen.looks_like_image_model(m): - found.append((key, m)) - return {"found": found} - - def done(res): - self._img_scan_worker = None - self._all_image_models = list(res.get("found", [])) - if getattr(self, "_pending_img_suggest", False): - self._pending_img_suggest = False - self._suggest_cross_provider_image() - - w = AgentWorker(job) - w.finished_ok.connect(done) - w.failed.connect(lambda _e: setattr(self, "_img_scan_worker", None)) - self._img_scan_worker = w - if then_suggest: - self._pending_img_suggest = True - w.start() - - def _maybe_suggest_image_model(self, instruction: str) -> None: - """If the request looks image-related, suggest a suitable image model - BEFORE running — searching the active provider first, then ALL providers. - The suggested model is what image generation will auto-use.""" - from ...core import image_gen - low = (instruction or "").lower() - if not any(w in low for w in self._IMAGE_WORDS): - return - picked = self.ai_model_combo.currentData() - if picked and image_gen.looks_like_image_model(picked): - return - local = image_gen.suggest_image_model(self._ai_models) - if local: - self.ai_chat.add_status(tr("folder.ai_image_suggest", model=local)) - return - # None on the active provider → look across ALL providers (cached, or scan - # now and suggest when the scan returns). - if self._all_image_models: - self._suggest_cross_provider_image() - elif self._img_scan_worker is not None: - self._pending_img_suggest = True # a scan is already running - else: - self._scan_all_image_models(then_suggest=True) - - def _suggest_cross_provider_image(self) -> None: - """Post a suggestion listing image models found on OTHER providers. When - none exist anywhere, fall back to telling the user their PICKED model - will be used for image generation (or that there's nothing to use).""" - from ...config import PROVIDER_LABELS - if not self._all_image_models: - picked = self.ai_model_combo.currentData() - if picked: - self.ai_chat.add_status(tr("folder.ai_image_use_selected", model=picked)) - else: - self.ai_chat.add_status(tr("folder.ai_image_none")) - return - seen, lines = set(), [] - for key, model in self._all_image_models: - tag = (key, model) - if tag in seen: - continue - seen.add(tag) - lines.append(f"• {model} ({PROVIDER_LABELS.get(key, key)})") - if len(lines) >= 5: - break - self.ai_chat.add_status(tr("folder.ai_image_suggest_all") + "\n" + "\n".join(lines)) diff --git a/presentation/graph/graph_project.py b/presentation/graph/graph_project.py deleted file mode 100644 index 4cb6c0a..0000000 --- a/presentation/graph/graph_project.py +++ /dev/null @@ -1,109 +0,0 @@ -"""Chọn project và đổi chế độ xem cho GraphRAG — R08-T14. - -Đồ thị luôn thuộc về một project. Đổi project là phải quét lại từ đầu, nên -phần này giữ luôn việc dọn kết quả cũ trước khi nạp cái mới. - -Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``. -""" -from __future__ import annotations - -from .graph_qa_widget import GraphQaMixin -from .graph_render import GraphRenderMixin -from .graph_scene import _Edge, _GraphView, _Node -import re -import sys -from pathlib import Path -from PySide6.QtCore import QPointF, Qt, QTimer, Signal -from PySide6.QtGui import QColor -from PySide6.QtWidgets import QComboBox, QFileDialog, QGraphicsScene, QGraphicsView, QHBoxLayout, QLabel, QLineEdit, QPushButton, QSplitter, QStackedWidget, QTabBar, QTextBrowser, QVBoxLayout, QWidget -from ...theme import current_palette -from ...core.worker import AgentWorker -from ...i18n import on_language_changed, tr -from ...state import AppContext -from ...ui.icons import collapse_right_icon, icon -from ...ui.widgets import CollapseStrip - - -class GraphProjectMixin: - """Chọn project + đổi tab xem. Trộn vào StructureGraphView.""" - - def _retranslate(self) -> None: - self.path_edit.setPlaceholderText(tr("structure.path_placeholder")) - self._pick_btn.setText(tr("structure.browse")) - self._scan_btn.setText(tr("structure.scan")) - self._export_btn.setText(tr("structure.export_png")) - # Both views are named at once now, so neither label depends on state. - self.view_tabs.setTabText(0, tr("structure.graph_btn")) - self.view_tabs.setTabText(1, tr("structure.msgs_btn")) - self.view_tabs.setTabToolTip(1, tr("structure.msgs_tooltip")) - self._ag_collapse.setToolTip(tr("structure.collapse_agent_tooltip")) - self._ag_label.setText(tr("structure.agent_header")) - self.ask_edit.setPlaceholderText(tr("structure.ask_placeholder")) - self._ask_btn.setText(tr("structure.ask")) - if self._detail_mode == "idle": - self.detail.setPlaceholderText(tr("structure.detail_placeholder")) - self._agent_strip.setToolTip(tr("structure.expand_agent_tooltip")) - self.project_combo.setToolTip(tr("structure.project_tooltip")) - self._refresh_project_combo() - def _refresh_project_combo(self) -> None: - from ...core.projects import list_projects - - keep = self._active_project_id - self.project_combo.blockSignals(True) - self.project_combo.clear() - self.project_combo.addItem(tr("structure.project_none"), "") - row_to_select = 0 - for i, p in enumerate(list_projects(), start=1): - self.project_combo.addItem(p.name, p.project_id) - if p.project_id == keep: - row_to_select = i - self.project_combo.setCurrentIndex(row_to_select) - self.project_combo.blockSignals(False) - def set_project(self, project_id: str) -> None: - pid = project_id or "" - self._refresh_project_combo() - target = self.project_combo.findData(pid) - if target < 0: - target = 0 - if self.project_combo.currentIndex() == target: - self._on_project_changed(target) - else: - self.project_combo.setCurrentIndex(target) - def _on_project_changed(self, _idx: int) -> None: - from ...core.projects import load_project - - pid = self.project_combo.currentData() or "" - project_changed = pid != self._active_project_id - if project_changed: - self._clear_extracts() # different workspace → drop temp extraction - self._active_project_id = pid - locked = bool(pid) - self.path_edit.setReadOnly(locked) - # Also disable the folder-pick button — otherwise the scan path is only - # "locked" against typing, but the picker could still repoint it outside - # the selected project's sandbox, breaking GraphRAG scope isolation. - self._pick_btn.setEnabled(not locked) - if locked: - project = load_project(pid) - if project is not None: - self.path_edit.setText(str(project.workspace_dir())) - if project_changed: - # Mark it and scan on the next visit rather than now. The rail's - # project picker made switching a one-click thing from any screen, - # and each switch rebuilt this graph — a folder walk plus a force - # layout plus a full setHtml of the D3 page — for a tab that was - # usually not even on screen. auto_scan_and_fit() picks the flag up - # when GraphRAG is actually opened. - self._needs_scan = True - def _pick(self) -> None: - chosen = QFileDialog.getExistingDirectory(self, tr("structure.pick_folder_title"), self.path_edit.text()) - if chosen: - self.path_edit.setText(chosen) - def _on_view_tab(self, index: int) -> None: - """Tab 0 = graph, tab 1 = messages. Same two views as before, now named - on screen instead of hidden behind one button's changing label.""" - if index == 1: - self._reload_messages() - self._stack.setCurrentWidget(self._msgs_view) - else: - self._stack.setCurrentWidget(self.web if self.web is not None else self.view) diff --git a/presentation/graph/graph_render.py b/presentation/graph/graph_render.py deleted file mode 100644 index de53cc6..0000000 --- a/presentation/graph/graph_render.py +++ /dev/null @@ -1,227 +0,0 @@ -"""Quét mã nguồn, dựng đồ thị, và xuất ảnh — R08-T14. - -Hai đường vẽ song song: khung nhìn Qt (``_render``) và bản D3 chạy trong -QtWebEngine (``_render_d3``). WebEngine nặng nên chỉ dựng ở lần hiện đầu tiên -(``_ensure_web``), và ``prewarm`` hâm nóng nó lúc rảnh để bấm vào GraphRAG -không phải ngồi nhìn khung trắng. - -Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``. -""" -from __future__ import annotations - -from .graph_scene import _Bridge, _Edge, _Node - -from .graph_web import _HAS_WEB, QWebChannel, QWebEngineView - -import math -import re -from pathlib import Path -from PySide6.QtCore import QPointF, Qt, QUrl -from PySide6.QtGui import QColor -from PySide6.QtWidgets import QFileDialog -from ...theme import current_palette -from ...core.worker import AgentWorker -from ...i18n import tr - - -class GraphRenderMixin: - """Quét, vẽ, xuất. Trộn vào StructureGraphView.""" - - def schedule_rescan(self, path: str = "") -> None: - if self._graph is None: - self._needs_scan = True - return - self._rescan_timer.start() - def prewarm(self) -> None: - """Pay for the graph view before it is clicked on, not during. - - Opening GraphRAG built a QWebEngineView (~140ms) and scanned the project - (~485ms) while an empty browser sat on screen — long enough, and white - enough, to read as the app restarting itself. Called from an idle timer - after the window is up, so startup itself is unaffected; the memory the - lazy construction was saving is spent a few seconds later instead. - """ - if not _HAS_WEB or self.web is not None: - return - self._ensure_web() - if self._graph is None and self.path_edit.text().strip(): - self._needs_scan = False - self._scan() # runs on a worker thread - def _ensure_web(self) -> None: - if self.web is not None or not _HAS_WEB: - return - self.web = QWebEngineView() - # Blank the page in the app's own background first. A fresh - # QWebEngineView paints white, and on a dark theme that white rectangle - # WAS the flash — it showed for as long as the first scan took. - self.web.setHtml( - f"") - self._bridge = _Bridge() - self._channel = QWebChannel() - self._channel.registerObject("py", self._bridge) - self.web.page().setWebChannel(self._channel) - self._stack.addWidget(self.web) - self._stack.setCurrentWidget(self.web) - if self._graph is not None: - self._render_d3() - def auto_scan_and_fit(self) -> None: - self._ensure_web() - if not self.path_edit.text().strip(): - return - if getattr(self, "_worker", None) is not None and self._worker.isRunning(): - self._fit() - self._preserve_answer() - return - if self._graph is not None and not self._needs_scan: - self._fit() - self._preserve_answer() - return - self._needs_scan = False - self._scan() - def _scan(self) -> None: - path = self.path_edit.text().strip() or str(Path.cwd()) - mode = "files" # default: scan all files (filter removed) - use_cmem = bool(self.ctx.config.codebase_memory.get("enabled")) - cmem_bin = self.ctx.config.codebase_memory.get("binary_path", "") - st = self.ctx.config.structure - max_nodes = int(st.get("max_nodes", 500) or 0) - max_edges = int(st.get("max_edges", 500) or 0) - self._scan_seq += 1 - seq = self._scan_seq - self.status_message.emit(tr("structure.scanning")) - - def job(worker: AgentWorker): - from ...core.structure_graph import ( - build_from_codebase_memory, build_from_directory, force_layout, - ) - if use_cmem: - from ...core.codebase_memory import CodebaseMemory - mem = CodebaseMemory(cmem_bin) - graph = (build_from_codebase_memory(mem, path, mode, max_nodes, max_edges) - if mem.available else build_from_directory(path, mode, max_nodes, max_edges)) - else: - graph = build_from_directory(path, mode, max_nodes, max_edges) - pos = force_layout(graph) - return {"graph": graph, "pos": pos, "seq": seq} - - w = AgentWorker(job) - w.finished_ok.connect(self._render) - w.failed.connect(lambda e: self.status_message.emit(tr("structure.scan_error", err=e))) - self._worker = w - w.start() - def _render(self, result: dict) -> None: - if result.get("seq") is not None and result["seq"] != self._scan_seq: - return - graph = result.get("graph") - pos = result.get("pos", {}) - if graph is None: - return - self._graph = graph - - self.scene.clear() - self.scene.setBackgroundBrush(QColor(current_palette().bg)) # restore after clear - self._node_items = [] - self._edge_items = [] - degree = {n.id: 0 for n in graph.nodes} - for e in graph.edges: - if e.source in degree: - degree[e.source] += 1 - if e.target in degree: - degree[e.target] += 1 - items = {} - sx = sy = 0.0 - for node in graph.nodes: - radius = int(8 + min(20, 2.2 * math.sqrt(degree.get(node.id, 0)))) - item = _Node(node, radius) - x, y = pos.get(node.id, (0, 0)) - item.setPos(x, y) - self.scene.addItem(item) - items[node.id] = item - self._node_items.append(item) - sx += x - sy += y - for edge in graph.edges: - a, b = items.get(edge.source), items.get(edge.target) - if a and b: - e = _Edge(a, b, getattr(edge, "type", "")) - self.scene.addItem(e) - self._edge_items.append(e) - n = max(1, len(self._node_items)) - self._centroid = QPointF(sx / n, sy / n) - self._fit() - - if self.web is not None: - self._render_d3() - - note = tr("structure.truncated_note") if getattr(graph, "truncated", False) else "" - self.status_message.emit(tr( - "structure.graph_summary", nodes=len(graph.nodes), edges=len(graph.edges), note=note)) - self._preserve_answer() - def _render_d3(self) -> None: - if self.web is None or self._graph is None: - return - from ...core.d3_graph import build_html - try: - self.web.setHtml(build_html(self._graph), QUrl("https://cowork.local/")) - except Exception as exc: - self.status_message.emit(f"D3 view error: {exc}") - def _on_selection(self) -> None: - for item in self.scene.selectedItems(): - if isinstance(item, _Node): - d = item.data - self.detail.setPlainText(f"[{d.kind.upper()}] {d.label}\n\n{d.detail}") - self._detail_mode = "node" - return - def _fit(self) -> None: - if self.web is not None and self._stack.currentWidget() is self.web: - self.web.page().runJavaScript("window.fitGraph && window.fitGraph();") - return - rect = self.scene.itemsBoundingRect() - if not rect.isNull(): - self.view.fitInView(rect.adjusted(-40, -40, 40, 40), Qt.KeepAspectRatio) - def _export(self) -> None: - path, _ = QFileDialog.getSaveFileName( - self, tr("structure.export_title"), "structure-graph.png", "PNG (*.png)") - if not path: - return - showing_d3 = (self.web is not None and self._stack.currentWidget() is self.web) - if showing_d3: - self._export_d3_png(path) - else: - self._export_widget_grab(path) - def _export_d3_png(self, path: str) -> None: - def on_result(data_url) -> None: - if not isinstance(data_url, str) or "," not in data_url: - self._export_widget_grab(path) - return - import base64 - try: - with open(path, "wb") as f: - f.write(base64.b64decode(data_url.split(",", 1)[1])) - self.status_message.emit(tr("structure.export_done", path=path)) - except (OSError, ValueError) as exc: - self.status_message.emit(tr("structure.export_failed", err=str(exc))) - self.web.page().runJavaScript("window.exportPng ? window.exportPng() : ''", on_result) - def _export_widget_grab(self, path: str) -> None: - ok = self._stack.currentWidget().grab().save(path, "PNG") - if ok: - self.status_message.emit(tr("structure.export_done", path=path)) - else: - self.status_message.emit(tr("structure.export_failed", err="grab() returned no image")) - @staticmethod - def _graph_context(graph) -> str: - from collections import defaultdict - by_kind = defaultdict(list) - for n in graph.nodes: - by_kind[n.kind].append(n.label) - lines = [] - for kind in ("file", "class", "function", "method", "module", "section"): - items = by_kind.get(kind, []) - if items: - lines.append(f"{kind} ({len(items)}): " + ", ".join(items[:60])) - id2label = {n.id: n.label for n in graph.nodes} - rels = [f"{id2label.get(e.source, e.source)} -{e.type}-> {id2label.get(e.target, e.target)}" - for e in graph.edges[:140]] - if rels: - lines.append("Relationships (sample):\n" + "\n".join(rels)) - return "\n".join(lines)[:7000] diff --git a/presentation/graph/graph_scene.py b/presentation/graph/graph_scene.py deleted file mode 100644 index 138c3aa..0000000 --- a/presentation/graph/graph_scene.py +++ /dev/null @@ -1,138 +0,0 @@ -"""Các phần tử vẽ của đồ thị: node, cạnh, khung nhìn — R08-T14. - -Thuần đồ hoạ Qt, không biết gì về GraphRAG hay agent. Tách riêng vì đây là -chỗ duy nhất cần mở khi chỉnh cách đồ thị trông ra sao — màu, hình mũi tên, -cách kéo thả và phóng to. -""" -from __future__ import annotations - -import re -from pathlib import Path -from PySide6.QtCore import QObject, QPointF, Qt, Slot -from PySide6.QtGui import QBrush, QColor, QFont, QPen -from PySide6.QtWidgets import QGraphicsEllipseItem, QGraphicsLineItem, QGraphicsSimpleTextItem, QGraphicsView -from ...core.structure_graph import EDGE_KIND_COLORS, NODE_KIND_COLORS -from ...theme import current_palette -from ...i18n import tr -from ...ui.osutil import open_folder, open_location - - -class _Bridge(QObject): - """Exposed to the D3 page so a Shift+click on a node can open its - storage folder/link (local path or URL — see osutil.open_location).""" - - @Slot(str) - def openPath(self, path: str) -> None: # noqa: N802 - JS-facing name - if path: - open_location(path) - -class _Edge(QGraphicsLineItem): - def __init__(self, a: "_Node", b: "_Node", type_: str = ""): - super().__init__() - self.a, self.b = a, b - self.type = type_ - # Colour the edge by its RELATIONSHIP type (contains/defines/method/…), - # so the graph shows what each connection MEANS — falling back to the - # source node's tint for any untyped edge. - color = QColor(EDGE_KIND_COLORS.get(type_, "")) if type_ else QColor() - if not color.isValid(): - color = a.brush().color().lighter(130) - self._color = color - self.setPen(QPen(color, 1.4)) - self.setZValue(-1) - # A small label naming the relationship, shown at the edge midpoint. - self._label = None - if type_: - self._label = QGraphicsSimpleTextItem(type_, self) - self._label.setBrush(QBrush(color.lighter(140))) - f = QFont() - f.setPointSize(7) - self._label.setFont(f) - self._label.setZValue(0) - a.edges.append(self) - b.edges.append(self) - self.adjust() - - def adjust(self) -> None: - pa, pb = self.a.scenePos(), self.b.scenePos() - self.setLine(pa.x(), pa.y(), pb.x(), pb.y()) - if self._label is not None: - br = self._label.boundingRect() - self._label.setPos((pa.x() + pb.x()) / 2 - br.width() / 2, - (pa.y() + pb.y()) / 2 - br.height() / 2) - -class _Node(QGraphicsEllipseItem): - def __init__(self, data, radius: int): - super().__init__(-radius, -radius, 2 * radius, 2 * radius) - self.data = data - self.edges = [] - tok = current_palette() - # NODE_KIND_COLORS is a categorical data encoding (one hue per node - # kind), not UI chrome — it stays fixed across themes on purpose so a - # given kind is always the same colour. Only the chrome follows tokens. - color = QColor(NODE_KIND_COLORS.get(data.kind, tok.text_muted)) - self.setBrush(QBrush(color)) - self.setPen(QPen(color.darker(160), 1.5)) - self.setFlags( - QGraphicsEllipseItem.ItemIsMovable - | QGraphicsEllipseItem.ItemIsSelectable - | QGraphicsEllipseItem.ItemSendsGeometryChanges - ) - self.setZValue(1) - label = QGraphicsSimpleTextItem(data.label, self) - label.setBrush(QBrush(QColor(tok.text))) - label.setPos(radius + 3, -8) - - def itemChange(self, change, value): # noqa: N802 - if change == QGraphicsEllipseItem.ItemPositionHasChanged: - for edge in self.edges: - edge.adjust() - return super().itemChange(change, value) - -class _GraphView(QGraphicsView): - def __init__(self, scene): - super().__init__(scene) - self.setDragMode(QGraphicsView.NoDrag) - self._panning = False - self._pan_start = QPointF() - - def wheelEvent(self, e): # noqa: N802 - self.scale(1.15 if e.angleDelta().y() > 0 else 1 / 1.15, - 1.15 if e.angleDelta().y() > 0 else 1 / 1.15) - - def mousePressEvent(self, e): # noqa: N802 - if e.button() == Qt.LeftButton and self.itemAt(e.pos()) is None: - self._panning = True - self._pan_start = e.position() - self.setCursor(Qt.ClosedHandCursor) - e.accept() - return - super().mousePressEvent(e) - - def mouseMoveEvent(self, e): # noqa: N802 - if self._panning: - delta = e.position() - self._pan_start - self._pan_start = e.position() - self.horizontalScrollBar().setValue(int(self.horizontalScrollBar().value() - delta.x())) - self.verticalScrollBar().setValue(int(self.verticalScrollBar().value() - delta.y())) - e.accept() - return - super().mouseMoveEvent(e) - - def mouseReleaseEvent(self, e): # noqa: N802 - if self._panning: - self._panning = False - self.setCursor(Qt.ArrowCursor) - e.accept() - return - super().mouseReleaseEvent(e) - - def mouseDoubleClickEvent(self, e): # noqa: N802 - """Double-click or Ctrl+click on a node opens its storage folder.""" - item = self.itemAt(e.pos()) - if isinstance(item, _Node) and getattr(item.data, "path", ""): - open_folder(item.data.path) - e.accept() - return - super().mouseDoubleClickEvent(e) - diff --git a/presentation/graph/graph_web.py b/presentation/graph/graph_web.py deleted file mode 100644 index 0211088..0000000 --- a/presentation/graph/graph_web.py +++ /dev/null @@ -1,38 +0,0 @@ -"""Có dùng được QtWebEngine hay không — R08-T14. - -Cờ khả năng, tách riêng vì cả ``structure_graph_view.py`` lẫn -``graph_render.py`` đều phải hỏi. Để ở một trong hai thì file kia import -ngược lại — vòng import. - -WebEngine là add-on tuỳ chọn của PySide6, và bản đóng gói one-file của -PyInstaller không chạy được nó; những trường hợp đó rơi về khung nhìn Qt. -""" -from __future__ import annotations - -import sys -from pathlib import Path - -def _frozen_onefile() -> bool: - """True only for a PyInstaller ONEFILE build. Onefile extracts itself to a - temp dir (sys._MEIPASS under %TEMP%), where the QtWebEngine helper process - can't run — creating a QWebEngineView hard-crashes the app (reported as - "click the graph tab → app closes"). A ONEDIR build keeps _MEIPASS as the - ``_internal`` folder right next to the exe, where WebEngine works fine, so - it keeps the full embedded D3 view.""" - if not getattr(sys, "frozen", False): - return False - meipass = getattr(sys, "_MEIPASS", "") - if not meipass: - return False - try: - return Path(meipass).resolve().parent != Path(sys.executable).resolve().parent - except OSError: # can't tell → play safe: use the native fallback - return True - - -try: # WebEngine + WebChannel are optional PySide6 add-ons - from PySide6.QtWebEngineWidgets import QWebEngineView - from PySide6.QtWebChannel import QWebChannel - _HAS_WEB = not _frozen_onefile() -except Exception: # pragma: no cover - _HAS_WEB = False diff --git a/presentation/scheduling/task_actions.py b/presentation/scheduling/task_actions.py deleted file mode 100644 index 431e59c..0000000 --- a/presentation/scheduling/task_actions.py +++ /dev/null @@ -1,194 +0,0 @@ -"""Các thao tác trên một task: thêm, sửa, chạy ngay, xoá, xem log — R08-T11. - -Tách khỏi ``ScheduleTaskTab`` để phần dựng bảng và phần hành động không nằm -lẫn nhau. ``_context_menu`` là chỗ tập trung: nó quyết định mục nào hiện ra -tuỳ theo đang chọn một hay nhiều thẻ. - -Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``. -""" -from __future__ import annotations - -# Import muộn trong hàm ở chỗ dùng: ba lớp này nằm cùng gói và một trong số -# chúng trộn ngược mixin này vào, nên import ở mức module là vòng. - -import copy -from pathlib import Path -from typing import Dict, List, Optional -from PySide6.QtCore import Qt, Signal -from PySide6.QtWidgets import ( - QAbstractItemView, QComboBox, QDialog, QDialogButtonBox, QHBoxLayout, - QLabel, QLineEdit, QListWidget, QListWidgetItem, QMenu, QMessageBox, - QPlainTextEdit, QPushButton, QScrollArea, QSizePolicy, QStackedWidget, - QTabBar, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, -) -from ...core import tasks as taskrepo -from ...core.projects import list_projects -from ...core.tasks import STATUSES, chain_error, duplicate_task, new_task -from ...core.worker import AgentWorker -from ...i18n import on_language_changed, tr -from ...state import AppContext -from ...theme import current_palette -from ...ui.calendar_view import CalendarView -from ...ui.icons import icon -from ...ui.osutil import open_path - - -class TaskActionsMixin: - """Thao tác trên task. Trộn vào ScheduleTaskTab.""" - - def _add_task_on_date(self, date_str: str) -> None: - """Create a task pre-filled with the clicked calendar date (default - 09:00) — same editor Add Task opens, nothing is saved until confirmed.""" - from ...ui.task_editor_dialog import TaskEditorDialog - - t = new_task("", schedule={"enabled": True, "run_at": f"{date_str} 09:00"}) - dlg = TaskEditorDialog(t, taskrepo.list_tasks(self._tasks_dir), self, ctx=self.ctx) - if dlg.exec() and dlg.edited_task: - self._save_and_refresh(dlg.edited_task) - self.status_message.emit(tr("schedtask.msg_created")) - def _save_and_refresh(self, task: dict) -> None: - taskrepo.save_task(task, self._tasks_dir) - self.refresh() - def _add_task(self) -> None: - from ...ui.task_editor_dialog import TaskEditorDialog - - dlg = TaskEditorDialog(None, taskrepo.list_tasks(self._tasks_dir), self, ctx=self.ctx) - if dlg.exec() and dlg.edited_task: - self._save_and_refresh(dlg.edited_task) - self.status_message.emit(tr("schedtask.msg_created")) - def _edit_task(self, task_id: str) -> None: - from ...ui.task_editor_dialog import TaskEditorDialog - - task = taskrepo.load_task(task_id, self._tasks_dir) - if not task: - return - dlg = TaskEditorDialog(task, taskrepo.list_tasks(self._tasks_dir), self, ctx=self.ctx) - if dlg.exec() and dlg.edited_task: - self._save_and_refresh(dlg.edited_task) - def _on_double_click(self, item: QListWidgetItem) -> None: - tid = item.data(Qt.UserRole) - if tid: - self._edit_task(tid) - def _is_multi_selection(item, selected) -> bool: - """True when the right-clicked card is part of an existing multi-item - selection — pure boolean, kept separate from _context_menu so it's - testable without ever invoking Qt's (modal, event-loop-blocking) menu.""" - return len(selected) > 1 and item in selected - def _context_menu(self, col: _KanbanColumn, pos) -> None: - item = col.itemAt(pos) - if item is None or not item.data(Qt.UserRole): - return - selected = [it for it in col.selectedItems() if it.data(Qt.UserRole)] - if self._is_multi_selection(item, selected): - self._bulk_delete_menu(col, pos, selected) - return - tid = item.data(Qt.UserRole) - task = taskrepo.load_task(tid, self._tasks_dir) - if not task: - return - menu = QMenu(col) - run_act = menu.addAction(tr("schedtask.menu_run")) - edit_act = menu.addAction(tr("schedtask.menu_edit")) - dup_act = menu.addAction(tr("schedtask.menu_duplicate")) - paused = task.get("status") == "paused" - pause_act = menu.addAction(tr("schedtask.menu_resume" if paused else "schedtask.menu_pause")) - logs_act = menu.addAction(tr("schedtask.menu_logs")) - hist_act = menu.addAction(tr("schedtask.menu_history")) - next_act = menu.addAction(tr("schedtask.menu_create_next")) - menu.addSeparator() - del_act = menu.addAction(tr("schedtask.menu_delete")) - chosen = menu.exec(col.viewport().mapToGlobal(pos)) - if chosen == run_act: - self._run_now(task) - elif chosen == edit_act: - self._edit_task(tid) - elif chosen == dup_act: - self._save_and_refresh(duplicate_task(task)) - elif chosen == pause_act: - task["status"] = "backlog" if paused else "paused" - self._save_and_refresh(task) - elif chosen == logs_act: - self._view_logs(task) - elif chosen == hist_act: - from .run_history_dialog import _RunHistoryDialog - _RunHistoryDialog(task, self).exec() - elif chosen == next_act: - self._create_next_from_output(task) - elif chosen == del_act: - if QMessageBox.question(self, tr("schedtask.menu_delete"), - tr("schedtask.delete_confirm", title=task.get("title", "")) - ) == QMessageBox.Yes: - taskrepo.delete_task(tid, self._tasks_dir) - self.refresh() - def _bulk_delete_menu(self, col: _KanbanColumn, pos, selected) -> None: - """Right-click on a multi-selection within one column (Shift/Ctrl-click - several cards first): one action deletes every selected task. The - popup itself is a thin wrapper — see _confirm_and_delete_selected for - the actual (independently testable) confirm+delete logic.""" - menu = QMenu(col) - del_act = menu.addAction(tr("schedtask.menu_delete_selected", n=len(selected))) - chosen = menu.exec(col.viewport().mapToGlobal(pos)) - if chosen == del_act: - self._confirm_and_delete_selected(selected) - def _confirm_and_delete_selected(self, selected) -> bool: - """Confirm, then delete every task in ``selected``. Split out of - _bulk_delete_menu so tests can drive it directly without having to - fake a real (modal, event-loop-blocking) QMenu popup.""" - if QMessageBox.question( - self, tr("schedtask.menu_delete"), - tr("schedtask.delete_multi_confirm", n=len(selected))) != QMessageBox.Yes: - return False - for item in selected: - tid = item.data(Qt.UserRole) - if tid: - taskrepo.delete_task(tid, self._tasks_dir) - self.refresh() - return True - def _run_now(self, task: dict) -> None: - if task.get("task_type") == "manual": - self.status_message.emit(tr("schedtask.msg_manual_norun")) - return - if self.scheduler is None: - self.status_message.emit(tr("schedtask.msg_no_scheduler")) - return - if self.scheduler.run_now(task["task_id"]): - self.status_message.emit(tr("schedtask.msg_running", title=task.get("title", ""))) - self.refresh() - def _view_logs(self, task: dict) -> None: - run_id = task.get("logs", {}).get("last_run_id") - if not run_id: - QMessageBox.information(self, tr("schedtask.menu_logs"), tr("schedtask.no_runs_yet")) - return - folder = taskrepo.ARTIFACTS_DIR / task["task_id"] / run_id - if folder.exists(): - open_path(str(folder)) - else: - QMessageBox.information(self, tr("schedtask.menu_logs"), tr("schedtask.no_runs_yet")) - def _create_next_from_output(self, task: dict) -> None: - """Scaffold a follow-up task pre-wired to consume this task's output.""" - nxt = new_task(tr("schedtask.next_of", title=task.get("title", ""))) - nxt["task_type"] = "cowork" - nxt["input"]["mode"] = "previous_task_output" - nxt["input"]["previous_task_id"] = task["task_id"] - nxt["dependency"]["previous_task_id"] = task["task_id"] - err = chain_error(taskrepo.list_tasks(self._tasks_dir) + [nxt], - task["task_id"], nxt["task_id"]) - if err: - QMessageBox.warning(self, tr("schedtask.g_dependency"), err) - return - taskrepo.save_task(nxt, self._tasks_dir) - task["dependency"]["next_task_id"] = nxt["task_id"] - task["dependency"]["pass_output_to_next"] = True - if task["dependency"].get("run_next_mode", "none") == "none": - task["dependency"]["run_next_mode"] = "run_after_success" - taskrepo.save_task(task, self._tasks_dir) - self.refresh() - self._edit_task(nxt["task_id"]) - def _ai_create(self) -> None: - from .ai_task_creator_dialog import _AiCreateDialog - dlg = _AiCreateDialog(self.ctx, self) - if dlg.exec() and dlg.created_tasks: - for t in dlg.created_tasks: - taskrepo.save_task(t, self._tasks_dir) - self.refresh() - self.status_message.emit(tr("schedtask.msg_ai_created", n=len(dlg.created_tasks)))