furk4ngg/capture-ai
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
28c7fca5f59c08aa116c095bb03e36bd74f53242
capture-ai/ai.py
T
FURK4NGG 14543379db Add files via upload
2026-05-21 17:36:55 +03:00

3809 lines
118 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr /bin/env python3
import os
import sys
import json
import base64
import io
import re
import tempfile
import mimetypes
import subprocess
import shutil
import argparse
from pathlib import Path
from datetime import datetime, timezone
from typing import NoReturn, Any, Dict
from memory import (
load_chat_data,
save_chat_data,
get_chat_messages,
add_memory_chunk,
update_code_context_from_message,
build_memory_context,
should_update_summary,
make_simple_summary,
)
import requests
CONFIG_PATH = Path.home() / ".config" / "capture-ai" / "config.json"
CACHE_BASE = Path.home() / ".cache" / "capture-ai"
REFS_CACHE_DIR = CACHE_BASE / "refs"
GENERATED_FILES_DIR = CACHE_BASE / "generated_files"
GENERATED_FILES_DIR.mkdir(parents=True, exist_ok=True)
LANG_DIR = Path.home() / "capture-ai" / "language"
_LANG_CACHE = None
_LANG_CACHE_LANG = None
OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
PROMPT_BLOCK_KEYS = {
"copyable": "o_Prompt_Copyable",
"apply": "o_Prompt_Apply",
"file_create": "o_Prompt_File_Create",
"web_search": "o_Prompt_Web_Search",
"structured": "o_Prompt_Structured",
"code": "o_Prompt_Code",
}
PDF_RULE_KEYS = {
"pdf_image": "o_Prompt_PDF_Image_Return_Rule",
"pdf_text_only_to_docx": "o_Prompt_PDF_Text_Only_To_DOCX_Rule",
"pdf_mixed_separate": "o_Prompt_PDF_Mixed_Separate_Rule",
"pdf_mixed_text_only": "o_Prompt_PDF_Mixed_Text_Only_Rule",
}
def get_prompt_text(cfg: dict, key: str) -> str:
text = get_ui_text(cfg, key)
return "" if text == key else str(text).strip()
def build_global_system_prompt(selected_blocks, cfg: dict) -> str:
if not isinstance(selected_blocks, list):
selected_blocks = ["copyable"]
parts = []
base = get_prompt_text(cfg, "o_Base_System_Prompt")
if base:
parts.append(base)
for block_name in selected_blocks:
block_name = str(block_name or "").strip()
lang_key = PROMPT_BLOCK_KEYS.get(block_name)
if not lang_key:
continue
block_text = get_prompt_text(cfg, lang_key)
if block_text:
parts.append(block_text)
return "\n\n".join(parts).strip()
def load_config():
try:
if CONFIG_PATH.exists():
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f) or {}
except Exception as e:
print("config read error:", e)
return {}
def _read_ui_language_from_config() -> str:
try:
if CONFIG_PATH.exists():
raw = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) or {}
return str(raw.get("ui_language", "en")).strip().lower() or "en"
except Exception:
pass
return "en"
def _load_language_map(lang: str) -> dict:
global _LANG_CACHE, _LANG_CACHE_LANG
lang = str(lang or "").strip().lower() or "en"
if _LANG_CACHE is not None and _LANG_CACHE_LANG == lang:
return _LANG_CACHE
lang_path = LANG_DIR / f"{lang}.json"
fallback_path = LANG_DIR / "en.json"
data = {}
try:
if lang_path.exists():
raw = json.loads(lang_path.read_text(encoding="utf-8")) or {}
if isinstance(raw, dict):
data = raw
except Exception as e:
print(f"language file read error ({lang_path.name}):", e)
if not data:
try:
if fallback_path.exists():
raw = json.loads(fallback_path.read_text(encoding="utf-8")) or {}
if isinstance(raw, dict):
data = raw
except Exception as e:
print(f"fallback language file read error ({fallback_path.name}):", e)
if not isinstance(data, dict):
data = {}
_LANG_CACHE = data
_LANG_CACHE_LANG = lang
return _LANG_CACHE
def get_ui_text(cfg: dict, key: str, **kwargs) -> str:
lang = str((cfg or {}).get("ui_language", "")).strip().lower()
if not lang:
lang = _read_ui_language_from_config()
lang_map = _load_language_map(lang)
if not lang_map:
return key
text = str(lang_map.get(key, key))
try:
return text.format(**kwargs)
except Exception:
return text
# ---------------------WEB SEARCH---------------------
def _tavily_search(cfg: dict, query: str, max_results: int = 5) -> str:
api_key = str(cfg.get("tavily_api_key") or "").strip()
if not api_key:
return (
"[WEB_SEARCH_ERROR]\n"
f"{get_ui_text(cfg, 'o_Tavily_API_Key_Missing')}\n"
"[/WEB_SEARCH_ERROR]"
)
query = str(query or "").strip()
if not query:
return (
"[WEB_SEARCH_ERROR]\n"
f"{get_ui_text(cfg, 'o_Empty_Search_Query')}\n"
"[/WEB_SEARCH_ERROR]"
)
try:
r = requests.post(
"https://api.tavily.com/search",
headers={
"Content-Type": "application/json"
},
json={
"api_key": api_key,
"query": query,
"search_depth": "basic",
"max_results": max_results,
"include_answer": False,
"include_raw_content": False
},
timeout=45
)
if r.status_code != 200:
try:
err_json = r.json()
err_text = (
err_json.get("detail")
or err_json.get("error")
or err_json.get("message")
or ""
)
except Exception:
err_text = ""
if r.status_code in (401, 403):
msg = get_ui_text(
cfg,
"o_Web_Search_API_Key_Invalid"
)
else:
msg = get_ui_text(
cfg,
"o_Web_Search_API_Error",
status_code=r.status_code
)
if err_text:
msg += f"\n{err_text}"
return (
"[WEB_SEARCH_ERROR]\n"
f"{msg}\n"
"[/WEB_SEARCH_ERROR]"
)
data = r.json()
results = data.get("results") or []
lines = ["[WEB_SEARCH_RESULTS]", f"query: {query}"]
for i, item in enumerate(results[:max_results], start=1):
title = str(item.get("title") or "").strip()
url = str(item.get("url") or "").strip()
content = str(item.get("content") or "").strip()
lines.append(f"\n{i}. {title}")
lines.append(f"url: {url}")
lines.append(f"snippet: {content}")
lines.append("[/WEB_SEARCH_RESULTS]")
return "\n".join(lines)
except Exception as e:
return f"[WEB_SEARCH_ERROR]\n{e}\n[/WEB_SEARCH_ERROR]"
def _extract_web_search_call(text: str) -> str | None:
raw = str(text or "").strip()
try:
data = json.loads(raw)
except Exception:
m = re.search(r"\{.*?\"tool\"\s*:\s*\"web_search\".*?\}", raw, re.S)
if not m:
return None
try:
data = json.loads(m.group(0))
except Exception:
return None
if not isinstance(data, dict):
return None
if str(data.get("tool") or "").strip() != "web_search":
return None
query = str(data.get("query") or "").strip()
return query or None
def _safe_read_text(path: Path, max_bytes: int = 250_000) -> str:
try:
data = path.read_bytes()
if len(data) > max_bytes:
data = data[:max_bytes]
return data.decode("utf-8", errors="replace")
except Exception as e:
return f"[File could not be read: {e}]"
def _safe_read_pdf(path: Path, max_chars: int = 50000) -> str:
try:
from pypdf import PdfReader
reader = PdfReader(str(path))
parts = []
for i, page in enumerate(reader.pages[:20], start=1):
txt = (page.extract_text() or "").strip()
if txt:
parts.append(f"\n--- PDF PAGE {i} ---\n{txt}")
return "\n".join(parts).strip()[:max_chars]
except Exception as e:
print("pdf text read error:", e, file=sys.stderr)
return ""
def _looks_like_empty_pdf_text(text: str) -> bool:
t = (text or "").strip()
return len(t) < 80
def _user_wants_image_edit(text: str) -> bool:
"""
Legacy auto-detection helper.
Şu an bilinçli olarak kullanılmıyor.
PDF kararları Prompt Chooser üzerinden veriliyor:
- pdf_text
- pdf_image
- pdf_text_image
PNG/JPG/WebP gibi normal görsellerde de model zaten görseli ve kullanıcı isteğini
birlikte gördüğü için ayrıca keyword tabanlı yönlendirme yapılmıyor.
İleride ayrı bir "Auto image edit mode" eklenirse bu fonksiyon yeniden
aktif kullanılabilir.
"""
return False
def _user_wants_text_edit(text: str) -> bool:
"""
Legacy auto-detection helper.
Şu an bilinçli olarak kullanılmıyor.
PDF text/image ayrımı artık Prompt Chooser tarafından belirleniyor.
İleride ayrı bir "Auto text edit mode" eklenirse bu fonksiyon yeniden
aktif kullanılabilir.
"""
return False
def _analyze_pdf_kind(path: Path) -> dict:
"""
PDF türünü ayırır:
- text_only
- image_only
- mixed
PyMuPDF çalışmazsa pypdf fallback kullanır.
"""
# 1) Önce PyMuPDF dene
try:
import fitz
doc = fitz.open(str(path))
text_parts = []
image_count = 0
for page in doc:
txt = (page.get_text("text") or "").strip()
if txt:
text_parts.append(txt)
image_count += len(page.get_images(full=True))
doc.close()
text = "\n\n".join(text_parts).strip()
has_text = len(text) >= 80
has_images = image_count > 0
if has_text and has_images:
kind = "mixed"
elif has_text:
kind = "text_only"
else:
kind = "image_only"
return {
"kind": kind,
"text": text,
"has_text": has_text,
"has_images": has_images,
"image_count": image_count,
}
except Exception as e:
print("pdf analyze fitz error:", e, file=sys.stderr)
# 2) Fallback: pypdf ile sadece text var mı bak
try:
text = _safe_read_pdf(path)
has_text = len((text or "").strip()) >= 80
if has_text:
# Görsel var mı kesin bilemeyiz, ama en azından image_only'e düşürmeyiz.
return {
"kind": "text_only",
"text": text,
"has_text": True,
"has_images": False,
"image_count": 0,
}
except Exception as e:
print("pdf analyze pypdf fallback error:", e, file=sys.stderr)
return {
"kind": "image_only",
"text": "",
"has_text": False,
"has_images": False,
"image_count": 0,
}
def _extract_pdf_mixed_layout(path: Path) -> dict:
"""
Mixed PDF için text ve image bloklarını koordinatlarıyla çıkarır.
Image bloklarını ayrıca dosya olarak kaydeder.
"""
try:
import fitz
doc = fitz.open(str(path))
image_dir = GENERATED_FILES_DIR / "pdf_layout_images" / path.stem
image_dir.mkdir(parents=True, exist_ok=True)
layout = {
"source_path": str(path.resolve()),
"pages": [],
}
for page_index, page in enumerate(doc):
page_dict = page.get_text("dict")
page_rect = page.rect
page_info = {
"page": page_index + 1,
"width": float(page_rect.width),
"height": float(page_rect.height),
"blocks": [],
}
for block_index, block in enumerate(page_dict.get("blocks", [])):
btype = block.get("type")
bbox = block.get("bbox")
if not bbox or len(bbox) != 4:
continue
block_id = f"p{page_index + 1}_b{block_index}"
# Text block
if btype == 0:
text_parts = []
for line in block.get("lines", []):
for span in line.get("spans", []):
t = str(span.get("text") or "")
if t.strip():
text_parts.append(t)
text = " ".join(text_parts).strip()
if text:
page_info["blocks"].append({
"id": block_id,
"type": "text",
"bbox": bbox,
"text": text,
})
# Image block
elif btype == 1:
image_bytes = block.get("image")
ext = str(block.get("ext") or "png").lower().strip() or "png"
if image_bytes:
img_path = image_dir / f"{block_id}.{ext}"
img_path.write_bytes(image_bytes)
page_info["blocks"].append({
"id": block_id,
"type": "image",
"bbox": bbox,
"path": str(img_path.resolve()),
})
layout["pages"].append(page_info)
doc.close()
return layout
except Exception as e:
print("mixed pdf layout extract error:", e, file=sys.stderr)
return {
"source_path": str(path),
"pages": [],
}
def _create_pdf_from_mixed_layout(
layout: dict,
text_replacements: dict | None,
image_replacements: dict | None,
output_name: str = "edited-mixed-pdf.pdf"
) -> Path | None:
"""
Mixed PDF'i yeniden kurar:
- Eski resimleri eski koordinatlarına koyar.
- AI'den gelen text_replacements ile textleri değiştirir.
"""
try:
import fitz
text_replacements = text_replacements if isinstance(text_replacements, dict) else {}
image_replacements = image_replacements if isinstance(image_replacements, dict) else {}
out = _generated_output_path(output_name, "pdf")
new_doc = fitz.open()
for page_info in layout.get("pages", []):
width = float(page_info.get("width") or 595)
height = float(page_info.get("height") or 842)
page = new_doc.new_page(width=width, height=height)
# Önce resimler
for block in page_info.get("blocks", []):
if block.get("type") != "image":
continue
bbox = block.get("bbox")
if not bbox or len(bbox) != 4:
continue
block_id = str(block.get("id") or "")
img_path = str(image_replacements.get(block_id) or block.get("path") or "")
if not img_path or not Path(img_path).exists():
continue
try:
page.insert_image(fitz.Rect(*bbox), filename=img_path)
except Exception as e:
print("mixed insert image error:", e, file=sys.stderr)
# Sonra textler
for block in page_info.get("blocks", []):
if block.get("type") != "text":
continue
bbox = block.get("bbox")
if not bbox or len(bbox) != 4:
continue
block_id = str(block.get("id") or "")
old_text = str(block.get("text") or "")
new_text = str(text_replacements.get(block_id, old_text) or "")
try:
page.insert_textbox(
fitz.Rect(*bbox),
new_text,
fontsize=11,
fontname="helv",
color=(0, 0, 0),
align=0,
)
except Exception as e:
print("mixed insert text error:", e, file=sys.stderr)
new_doc.save(str(out))
new_doc.close()
return out if out.exists() else None
except Exception as e:
print("mixed pdf create error:", e, file=sys.stderr)
return None
def _pdf_to_page_images(path: Path, max_pages: int = 1, zoom: float = 1.5) -> list[Path]:
out_dir = Path(tempfile.gettempdir()) / "capture-ai-pdf-pages" / path.stem
out_dir.mkdir(parents=True, exist_ok=True)
# Eski cache kalıntılarını temizle
for old in out_dir.glob("page*"):
try:
old.unlink()
except Exception:
pass
errors = []
# 1) PyMuPDF / fitz ile dene
try:
import fitz
doc = fitz.open(str(path))
out = []
page_count = min(len(doc), max_pages)
if page_count <= 0:
errors.append("fitz: PDF has 0 pages")
else:
for i in range(page_count):
page = doc[i]
pix = page.get_pixmap(
matrix=fitz.Matrix(zoom, zoom),
alpha=False
)
img_path = out_dir / f"page_{i + 1}.png"
pix.save(str(img_path))
if img_path.exists() and img_path.stat().st_size > 0:
out.append(img_path)
else:
errors.append(f"fitz: page {i + 1} produced empty PNG")
doc.close()
if out:
return out
except Exception as e:
errors.append(f"fitz: {e}")
# 2) Fallback: poppler / pdftoppm ile dene
try:
pdftoppm = shutil.which("pdftoppm")
if not pdftoppm:
errors.append("pdftoppm: command not found")
else:
prefix = out_dir / "page"
cmd = [
pdftoppm,
"-png",
"-r", "150",
"-f", "1",
"-l", str(max_pages),
str(path),
str(prefix),
]
proc = subprocess.run(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True
)
if proc.returncode != 0:
errors.append(f"pdftoppm: {proc.stderr.strip()}")
else:
out = sorted(out_dir.glob("page-*.png"))
out = [
p for p in out
if p.exists() and p.stat().st_size > 0
]
if out:
return out[:max_pages]
errors.append("pdftoppm: no PNG output")
except Exception as e:
errors.append(f"pdftoppm: {e}")
print("PDF_CONVERT_FAILED:", " | ".join(errors), file=sys.stderr)
return []
def _safe_read_docx(path: Path, max_chars: int = 50000) -> str:
try:
import docx
doc = docx.Document(str(path))
parts = []
for p in doc.paragraphs:
t = (p.text or "").strip()
if t:
parts.append(t)
# tabloları da oku
for table in doc.tables:
for row in table.rows:
cells = []
for cell in row.cells:
txt = (cell.text or "").strip()
if txt:
cells.append(txt)
if cells:
parts.append(" | ".join(cells))
out = "\n".join(parts).strip()
return out[:max_chars] if out else "[DOCX file is empty or text could not be extracted.]"
except Exception as e:
return f"[DOCX could not be read: {e}]"
def _safe_read_xlsx(path: Path, max_rows: int = 200, max_chars: int = 50000) -> str:
try:
from openpyxl import load_workbook
wb = load_workbook(str(path), data_only=True)
parts = []
for ws in wb.worksheets:
parts.append(f"\n--- SHEET: {ws.title} ---")
row_count = 0
for row in ws.iter_rows(values_only=True):
vals = ["" if v is None else str(v) for v in row]
if any(v.strip() for v in vals):
parts.append(" | ".join(vals))
row_count += 1
if row_count >= max_rows:
parts.append("[Sheet truncated]")
break
return "\n".join(parts).strip()[:max_chars]
except Exception as e:
return f"[XLSX could not be read: {e}]"
def _is_text_file(path: Path) -> bool:
ext = path.suffix.lower()
return ext in {
".txt", ".md", ".py", ".json", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".conf",
".sh", ".bash", ".zsh", ".js", ".ts", ".tsx", ".jsx", ".css", ".html", ".xml",
".c", ".cpp", ".h", ".hpp", ".rs", ".go", ".java", ".kt", ".cs", ".sql"
}
def _die(msg: str, code: int = 1) -> NoReturn:
sys.stderr.write(msg + "\n")
raise SystemExit(code)
def _load_config() -> Dict[str, Any]:
if not CONFIG_PATH.exists():
_die(get_ui_text({}, "o_Config_Not_Found"), 1)
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f) or {}
except Exception as e:
_die(get_ui_text({}, "o_Config_Read_Error", error=e), 1)
def _save_config(cfg: dict):
try:
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
CONFIG_PATH.write_text(json.dumps(cfg, ensure_ascii=False, indent=2), encoding="utf-8")
except Exception:
# config yazılamasa bile AI çalışsın
pass
def _parse_selected_indexes(arg: str) -> list[int]:
out = []
if not arg:
return out
for x in arg.split(","):
x = x.strip()
if x.isdigit():
out.append(int(x))
# uniq + stable
seen = set()
uniq = []
for i in out:
if i not in seen:
seen.add(i)
uniq.append(i)
return uniq
def _mime_for(path: Path) -> str:
mime, _ = mimetypes.guess_type(str(path))
return mime or "application/octet-stream"
def _data_url_for_image(path: Path) -> str:
mime = _mime_for(path)
b64 = base64.b64encode(path.read_bytes()).decode("ascii")
return f"data:{mime};base64,{b64}"
def _role_map(chat_role: str) -> str:
return "assistant" if chat_role == "bot" else "user"
def _safe_chat_key(chat_file: Path) -> str:
return chat_file.stem.replace("/", "_").replace("\\", "_")
def _utc_stamp() -> str:
return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
def _copy_to_cache(src: Path, dest_dir: Path) -> Path:
dest_dir.mkdir(parents=True, exist_ok=True)
base = src.stem.replace(" ", "_")
ext = src.suffix or ".img"
dest = dest_dir / f"{base}{ext}"
if dest.exists():
dest = dest_dir / f"{base}_{_utc_stamp()}{ext}"
shutil.copy2(src, dest)
return dest
def _extract_named_block(text: str, name: str) -> str | None:
pattern = rf"(?ms)^\s*{re.escape(name)}\s*\n(.*?)\n\s*{re.escape(name)}\s*$"
m = re.search(pattern, text or "")
return m.group(1).strip() if m else None
def _remove_named_block(text: str, name: str) -> str:
pattern = rf"(?ms)\n?\s*{re.escape(name)}\s*\n.*?\n\s*{re.escape(name)}\s*"
return re.sub(pattern, "", text or "").strip()
def _safe_output_name(name: str, fallback_ext: str) -> str:
name = str(name or "").strip()
if not name:
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
name = f"generated-{stamp}.{fallback_ext}"
for ch in ["/", "\\", "\n", "\r", "\t"]:
name = name.replace(ch, "_")
if "." not in Path(name).name:
name += f".{fallback_ext}"
return name
def _generated_output_path(output_name: str, fallback_ext: str) -> Path:
GENERATED_FILES_DIR.mkdir(parents=True, exist_ok=True)
safe_name = _safe_output_name(output_name, fallback_ext)
out = GENERATED_FILES_DIR / safe_name
if out.exists():
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
out = GENERATED_FILES_DIR / f"{out.stem}-{stamp}{out.suffix}"
return out
def _message_content_with_generated_files(msg: dict) -> str:
text = str(msg.get("content") or "").strip()
lines = []
if text:
lines.append(text)
images = msg.get("images")
if isinstance(images, list) and images:
lines.append("\n[GENERATED_IMAGES]")
for p in images:
if p:
lines.append(f"path: {p}")
lines.append("[/GENERATED_IMAGES]")
generated = msg.get("generated_files")
if isinstance(generated, list) and generated:
lines.append("\n[GENERATED_FILES]")
for f in generated:
if not isinstance(f, dict):
continue
path = str(f.get("path") or "").strip()
name = str(f.get("name") or Path(path).name).strip()
if path:
lines.append(f"name: {name}")
lines.append(f"path: {path}")
lines.append("[/GENERATED_FILES]")
return "\n".join(lines).strip()
def _sanitize_plain_document_text(content: str) -> str:
text = str(content or "")
# Markdown image: ![alt](url)
text = re.sub(r"!\[[^\]]*\]\([^)]+\)", "", text)
# HTML img tag
text = re.sub(r"<\s*img[^>]*>", "", text, flags=re.IGNORECASE)
# data:image base64 blokları
text = re.sub(r"data:image\/[a-zA-Z0-9.+-]+;base64,[A-Za-z0-9+/=\s]+", "", text)
# Basit HTML tag temizliği
text = re.sub(r"<[^>]+>", "", text)
# Çoklu boş satırları sadeleştir
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
def _create_docx_file(output_name: str, content: str) -> Path | None:
try:
import docx
import docx.shared
out = _generated_output_path(output_name, "docx")
doc = docx.Document()
style = doc.styles["Normal"]
style.font.name = "Arial"
style.font.size = docx.shared.Pt(12)
section = doc.sections[0]
section.top_margin = docx.shared.Cm(2)
section.bottom_margin = docx.shared.Cm(2)
section.left_margin = docx.shared.Cm(2)
section.right_margin = docx.shared.Cm(2)
content = _sanitize_plain_document_text(content)
for line in str(content or "").splitlines():
doc.add_paragraph(line)
doc.save(str(out))
return out
except Exception as e:
print("docx create error:", e, file=sys.stderr)
return None
def _convert_docx_to_pdf(docx_path: Path) -> Path | None:
"""
DOCX dosyasını LibreOffice ile PDF'e çevirir.
"""
try:
if not docx_path.exists():
return None
GENERATED_FILES_DIR.mkdir(parents=True, exist_ok=True)
subprocess.run(
[
"libreoffice",
"--headless",
"--convert-to",
"pdf",
"--outdir",
str(GENERATED_FILES_DIR),
str(docx_path),
],
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
converted = GENERATED_FILES_DIR / f"{docx_path.stem}.pdf"
if converted.exists():
return converted
return None
except Exception as e:
print("docx to pdf convert error:", e, file=sys.stderr)
return None
def _norm_filter_text(value: str) -> str:
import unicodedata
s = str(value or "").strip().lower()
s = unicodedata.normalize("NFKD", s)
s = "".join(ch for ch in s if not unicodedata.combining(ch))
tr_map = str.maketrans({
"ı": "i",
"ğ": "g",
"ü": "u",
"ş": "s",
"ö": "o",
"ç": "c",
})
return s.translate(tr_map)
def _xlsx_filter_keep_rows(source_path: str, output_name: str, spec: dict) -> Path | None:
try:
from openpyxl import load_workbook
src = Path(source_path).expanduser()
if not src.exists() or src.suffix.lower() != ".xlsx":
return None
out = _generated_output_path(output_name, "xlsx")
wb = load_workbook(str(src))
preferred_columns = spec.get("preferred_columns") or []
patterns = spec.get("patterns_keep_any") or []
preferred_columns_norm = [
_norm_filter_text(x) for x in preferred_columns
]
patterns_norm = [
_norm_filter_text(x) for x in patterns
]
fallback_all = bool(spec.get("fallback_search_all_columns", True))
for ws in wb.worksheets:
if ws.max_row < 2:
continue
headers = []
for cell in ws[1]:
headers.append(_norm_filter_text(cell.value))
target_indexes = []
for idx, header in enumerate(headers, start=1):
if header in preferred_columns_norm:
target_indexes.append(idx)
if not target_indexes and fallback_all:
target_indexes = list(range(1, ws.max_column + 1))
rows_to_delete = []
for row_idx in range(2, ws.max_row + 1):
values = []
for col_idx in target_indexes:
values.append(ws.cell(row=row_idx, column=col_idx).value)
joined = _norm_filter_text(" ".join(str(v or "") for v in values))
keep = any(pat in joined for pat in patterns_norm)
if not keep:
rows_to_delete.append(row_idx)
for row_idx in reversed(rows_to_delete):
ws.delete_rows(row_idx, 1)
wb.save(str(out))
return out
except Exception as e:
print("xlsx filter error:", e, file=sys.stderr)
return None
def _normalize_xlsx_rows(content):
rows = content
# Eğer string olarak Python dict/JSON geldiyse çöz
if isinstance(rows, str):
s = rows.strip()
try:
rows = json.loads(s)
except Exception:
try:
import ast
rows = ast.literal_eval(s)
except Exception:
return [[rows]]
# {"sheet": "People", "data": [{...}, {...}]}
# {"sheet_name": "People", "columns": [...], "rows": [...]}
# {"sheets": [{"data": [...]}]}
if isinstance(rows, dict):
if isinstance(rows.get("sheets"), list):
first_sheet = rows["sheets"][0] if rows["sheets"] else {}
if isinstance(first_sheet, dict):
rows = first_sheet.get("data", [])
elif isinstance(rows.get("data"), list):
data_rows = rows.get("data") or []
# data: [{Name:..., Age:...}, {...}]
if data_rows and all(isinstance(x, dict) for x in data_rows):
headers = list(data_rows[0].keys())
rows = [headers] + [
[item.get(h, "") for h in headers]
for item in data_rows
]
else:
rows = data_rows
elif isinstance(rows.get("rows"), list):
data_rows = rows.get("rows") or []
cols = rows.get("columns")
if isinstance(cols, list):
rows = [cols] + data_rows
else:
rows = data_rows
else:
return [[str(rows)]]
# [{"sheet": "...", "data": [...]}] gibi liste içinde dict geldiyse
if isinstance(rows, list) and rows and isinstance(rows[0], dict):
first = rows[0]
if isinstance(first.get("sheets"), list):
first_sheet = first["sheets"][0] if first["sheets"] else {}
if isinstance(first_sheet, dict):
rows = first_sheet.get("data", [])
elif isinstance(first.get("data"), list):
data_rows = first.get("data") or []
if data_rows and all(isinstance(x, dict) for x in data_rows):
headers = list(data_rows[0].keys())
rows = [headers] + [
[item.get(h, "") for h in headers]
for item in data_rows
]
else:
rows = data_rows
elif isinstance(first.get("rows"), list):
data_rows = first.get("rows") or []
cols = first.get("columns")
if isinstance(cols, list):
rows = [cols] + data_rows
else:
rows = data_rows
if not isinstance(rows, list):
return [[str(rows)]]
# [[["Name","Age"],["Ali",20]]] gibi tek satır içine gömülü tablo
if (
len(rows) == 1
and isinstance(rows[0], list)
and rows[0]
and all(isinstance(x, list) for x in rows[0])
):
rows = rows[0]
fixed_rows = []
for row in rows:
if isinstance(row, list):
fixed_rows.append(row)
elif isinstance(row, dict):
headers = list(row.keys())
if not fixed_rows:
fixed_rows.append(headers)
fixed_rows.append([row.get(h, "") for h in headers])
else:
fixed_rows.append([row])
return fixed_rows
def _create_xlsx_file(output_name: str, content) -> Path | None:
try:
from openpyxl import Workbook
out = _generated_output_path(output_name, "xlsx")
wb = Workbook()
ws = wb.active
ws.title = "Sheet1"
rows = _normalize_xlsx_rows(content)
for row in rows:
ws.append(row)
wb.save(str(out))
return out
except Exception as e:
print("xlsx create error:", e, file=sys.stderr)
return None
def _create_txt_like_file(output_name: str, content: str, ext: str) -> Path | None:
try:
out = _generated_output_path(output_name, ext)
out.write_text(str(content or ""), encoding="utf-8")
return out
except Exception as e:
print(f"{ext} create error:", e, file=sys.stderr)
return None
def _image_bytes_to_pdf_file(image_bytes: bytes, output_name: str = "edited.pdf") -> Path | None:
try:
from PIL import Image
out = _generated_output_path(output_name, "pdf")
img = Image.open(io.BytesIO(image_bytes))
if img.mode in ("RGBA", "LA", "P"):
img = img.convert("RGB")
elif img.mode != "RGB":
img = img.convert("RGB")
img.save(str(out), "PDF", resolution=100.0)
return out
except Exception as e:
print("image to pdf convert error:", e, file=sys.stderr)
return None
def _image_paths_to_pdf_file(image_paths: list[Path], output_name: str = "edited.pdf") -> Path | None:
try:
from PIL import Image
valid = [Path(p) for p in image_paths if p and Path(p).exists()]
if not valid:
return None
out = _generated_output_path(output_name, "pdf")
images = []
for p in valid:
img = Image.open(str(p))
if img.mode in ("RGBA", "LA", "P"):
img = img.convert("RGB")
elif img.mode != "RGB":
img = img.convert("RGB")
images.append(img)
if not images:
return None
first = images[0]
rest = images[1:]
first.save(str(out), "PDF", save_all=True, append_images=rest, resolution=100.0)
return out if out.exists() else None
except Exception as e:
print("images to pdf convert error:", e, file=sys.stderr)
return None
def _save_result_images_to_files(result_obj: dict) -> list[Path]:
saved = []
if not isinstance(result_obj, dict):
return saved
def save_raw(raw: bytes, ext: str = "png"):
out = _generated_output_path(f"edited-image.{ext}", ext)
out.write_bytes(raw)
if out.exists():
saved.append(out)
# URL images
for key in ("image", "url"):
val = result_obj.get(key)
if isinstance(val, str) and val.strip():
s = val.strip()
if s.startswith("data:image/") and "," in s:
try:
head, b64 = s.split(",", 1)
ext = head.split("/")[1].split(";")[0] or "png"
save_raw(base64.b64decode(b64), ext)
except Exception:
pass
elif s.startswith(("http://", "https://")):
try:
r = requests.get(s, timeout=120)
if r.status_code == 200:
save_raw(r.content, "png")
except Exception:
pass
imgs = result_obj.get("images")
if isinstance(imgs, list):
for item in imgs:
if isinstance(item, str):
s = item.strip()
if s.startswith("data:image/") and "," in s:
try:
head, b64 = s.split(",", 1)
ext = head.split("/")[1].split(";")[0] or "png"
save_raw(base64.b64decode(b64), ext)
except Exception:
pass
elif s.startswith(("http://", "https://")):
try:
r = requests.get(s, timeout=120)
if r.status_code == 200:
save_raw(r.content, "png")
except Exception:
pass
elif isinstance(item, dict):
u = item.get("url") or item.get("image_url")
if isinstance(u, dict):
u = u.get("url")
if isinstance(u, str) and u.strip():
try:
r = requests.get(u.strip(), timeout=120)
if r.status_code == 200:
save_raw(r.content, "png")
except Exception:
pass
# base64 images
for key in ("image_base64",):
b64 = result_obj.get(key)
if isinstance(b64, str) and b64.strip():
try:
save_raw(base64.b64decode(b64.split(",", 1)[-1]), "png")
except Exception:
pass
many_b64 = result_obj.get("images_base64")
if isinstance(many_b64, list):
for b64 in many_b64:
if isinstance(b64, str) and b64.strip():
try:
save_raw(base64.b64decode(b64.split(",", 1)[-1]), "png")
except Exception:
pass
return saved
def _overlay_images_on_pdf_blocks(source_pdf: Path, layout: dict, replacement_images: list[Path], output_name: str = "edited-pdf-image.pdf") -> Path | None:
try:
import fitz
source_pdf = Path(source_pdf)
if not source_pdf.exists():
return None
doc = fitz.open(str(source_pdf))
out = _generated_output_path(output_name, "pdf")
image_blocks = []
for page in layout.get("pages", []):
page_no = int(page.get("page") or 1)
for block in page.get("blocks", []):
if block.get("type") == "image":
image_blocks.append((page_no, block))
if not image_blocks or not replacement_images:
doc.close()
return None
for idx, img_path in enumerate(replacement_images):
if idx >= len(image_blocks):
break
page_no, block = image_blocks[idx]
bbox = block.get("bbox")
if not bbox or len(bbox) != 4:
continue
page = doc[page_no - 1]
page.insert_image(
fitz.Rect(*bbox),
filename=str(img_path),
overlay=True
)
doc.save(str(out))
doc.close()
return out if out.exists() else None
except Exception as e:
print("overlay pdf image error:", e, file=sys.stderr)
return None
def _save_ai_returned_file(name: str, data: bytes) -> Path | None:
try:
ext = Path(name or "").suffix.lower().lstrip(".") or "bin"
out = _generated_output_path(name, ext)
out.write_bytes(data)
return out
except Exception as e:
print("ai returned file save error:", e, file=sys.stderr)
return None
def _apply_ai_returned_files_from_result(result_obj: dict) -> list[dict]:
generated = []
if not isinstance(result_obj, dict):
return generated
# 1) Base64 dosyalar
files_b64 = result_obj.get("generated_files_base64")
if isinstance(files_b64, list):
for item in files_b64:
if not isinstance(item, dict):
continue
name = str(item.get("name") or item.get("output_name") or "generated.pdf").strip()
b64 = str(item.get("base64") or item.get("data") or "").strip()
if b64.startswith("data:") and "," in b64:
b64 = b64.split(",", 1)[1].strip()
if not b64:
continue
try:
raw = base64.b64decode(b64)
except Exception:
continue
out = _save_ai_returned_file(name, raw)
if out and out.exists():
generated.append({
"path": str(out.resolve()),
"name": out.name
})
# 2) URL dosyalar
files_url = result_obj.get("generated_files_url")
if isinstance(files_url, list):
for item in files_url:
if not isinstance(item, dict):
continue
name = str(item.get("name") or item.get("output_name") or "generated.pdf").strip()
url = str(item.get("url") or "").strip()
if not url.startswith(("http://", "https://")):
continue
try:
r = requests.get(url, timeout=120)
if r.status_code != 200:
continue
out = _save_ai_returned_file(name, r.content)
if out and out.exists():
generated.append({
"path": str(out.resolve()),
"name": out.name
})
except Exception as e:
print("ai returned file url download error:", e, file=sys.stderr)
return generated
def _apply_file_create_from_reply(reply_text: str) -> list[dict]:
block = _extract_named_block(reply_text, "file_create")
if not block:
return []
try:
ops = json.loads(block)
except Exception as e:
print("file_create json parse error:", e, file=sys.stderr)
return []
if isinstance(ops, dict):
ops = [ops]
if not isinstance(ops, list):
return []
generated = []
for op in ops:
if not isinstance(op, dict):
continue
fmt = str(op.get("format") or "").strip().lower().lstrip(".")
output_name = str(op.get("output_name") or "").strip()
content = op.get("content", "")
out = None
if fmt == "docx":
out = _create_docx_file(output_name, str(content or ""))
if out and out.exists():
generated.append({
"path": str(out.resolve()),
"name": out.name
})
pdf_out = _convert_docx_to_pdf(out)
if pdf_out and pdf_out.exists():
generated.append({
"path": str(pdf_out.resolve()),
"name": pdf_out.name
})
continue
elif fmt == "xlsx":
source_path = str(op.get("source_path") or "").strip()
if isinstance(content, list) and content and isinstance(content[0], dict):
first_spec = content[0]
action = str(first_spec.get("action") or "").strip()
if action == "filter_keep_rows":
out = _xlsx_filter_keep_rows(source_path, output_name, first_spec)
else:
out = _create_xlsx_file(output_name, content)
else:
out = _create_xlsx_file(output_name, content)
elif fmt in ("txt", "md"):
out = _create_txt_like_file(output_name, str(content or ""), fmt)
if out and out.exists():
generated.append({
"path": str(out.resolve()),
"name": out.name
})
return generated
def _build_blocks_and_cache_info(msg: dict, cache_images_dir: Path | None, pdf_mode: str = "auto"):
blocks = []
cached_image = None
# 1) Text
txt = msg.get("content") or ""
if txt:
blocks.append({"type": "text", "text": txt})
# 2) Images (legacy "image" + new "images")
imgs = []
one = msg.get("image")
if one:
imgs.append(one)
many = msg.get("images")
if isinstance(many, list):
imgs.extend([str(x) for x in many if x])
# uniq
seen = set()
uniq_imgs = []
for p in imgs:
if p not in seen:
seen.add(p)
uniq_imgs.append(p)
for img_raw in uniq_imgs:
img_path = Path(img_raw)
if not img_path.exists():
continue
if cache_images_dir is not None:
try:
cached_image = _copy_to_cache(img_path, cache_images_dir)
except Exception:
cached_image = cached_image # önceki varsa kalsın
try:
blocks.append({
"type": "image_url",
"image_url": {"url": _data_url_for_image(img_path)}
})
except Exception:
pass
# 3) Files (new: "files": [{"path","name","edit"}])
files = msg.get("files") or []
if isinstance(files, list):
for f in files:
if not isinstance(f, dict):
continue
p_raw = str(f.get("path") or "").strip()
if not p_raw:
continue
p = Path(p_raw)
if not p.exists() or not p.is_file():
continue
editable = bool(f.get("edit"))
name = (f.get("name") or p.name or "").strip()
mime = _mime_for(p)
file_text = None
file_type = "unsupported"
handled_file = False
if _is_text_file(p):
file_text = _safe_read_text(p)
file_type = "text"
elif p.suffix.lower() == ".pdf":
# IMAGE MODE override — HER ZAMAN image gönder
if pdf_mode == "image":
blocks.append({
"type": "text",
"text": (
"\n[FILE]\n"
f"name: {name}\n"
f"mime: {mime}\n"
f"editable: {'true' if editable else 'false'}\n"
"type: pdf_visual_pages\n"
"[/FILE]\n"
)
})
page_images = _pdf_to_page_images(p, max_pages=12, zoom=1.5)
if not page_images:
raise RuntimeError(
get_ui_text(load_config(), "o_PDF_Convert_Failed")
)
for img_path in page_images:
blocks.append({
"type": "image_url",
"image_url": {
"url": _data_url_for_image(img_path)
}
})
handled_file = True
continue
analysis = _analyze_pdf_kind(p)
pdf_kind = analysis.get("kind")
pdf_text = analysis.get("text") or ""
# 1) Sadece yazı PDF
if pdf_kind == "text_only":
file_text = pdf_text
file_type = "pdf_text_only"
if not str(file_text or "").strip():
blocks.append({
"type": "text",
"text": (
get_ui_text(load_config(), "o_PDF_Text_Extract_Failed")
)
})
handled_file = True
continue
# 2) Hem yazı hem resim PDF
elif pdf_kind == "mixed":
layout = _extract_pdf_mixed_layout(p)
if pdf_mode == "text":
text_blocks = []
for page in layout.get("pages", []):
for block in page.get("blocks", []):
if block.get("type") == "text":
text_blocks.append({
"id": block.get("id"),
"page": page.get("page"),
"text": block.get("text"),
})
blocks.append({
"type": "text",
"text": (
"\n[FILE]\n"
f"name: {name}\n"
f"source_path_for_app_only: {str(p)}\n"
f"mime: {mime}\n"
f"editable: {'true' if editable else 'false'}\n"
"type: pdf_mixed_text_only\n"
f"note: {get_ui_text(load_config(), 'o_File_Note_Images_Preserved')}\n"
"[CONTENT]\n"
f"{json.dumps(text_blocks, ensure_ascii=False)}\n"
"[/CONTENT]\n"
"[/FILE]\n"
)
})
handled_file = True
elif pdf_mode == "image":
image_blocks = []
for page in layout.get("pages", []):
for block in page.get("blocks", []):
if block.get("type") == "image":
image_blocks.append(block)
img_path = Path(str(block.get("path") or ""))
if img_path.exists():
blocks.append({
"type": "image_url",
"image_url": {
"url": _data_url_for_image(img_path)
}
})
if not image_blocks:
blocks.append({
"type": "text",
"text": get_ui_text(load_config(), "o_PDF_No_Images_Found")
})
handled_file = True
else:
blocks.append({
"type": "text",
"text": (
"\n[FILE]\n"
f"name: {name}\n"
f"path: {str(p)}\n"
f"mime: {mime}\n"
f"editable: {'true' if editable else 'false'}\n"
"type: pdf_mixed_separated_layout\n"
"[CONTENT]\n"
f"{json.dumps(layout, ensure_ascii=False)}\n"
"[/CONTENT]\n"
"[/FILE]\n"
)
})
handled_file = True
# 3) Yazı yok / scan / image PDF
else:
blocks.append({
"type": "text",
"text": (
"\n[FILE]\n"
f"name: {name}\n"
f"path: {str(p)}\n"
f"mime: {mime}\n"
f"editable: {'true' if editable else 'false'}\n"
"type: pdf_visual_pages\n"
f"note: {get_ui_text(load_config(), 'o_File_Note_PDF_Pages_As_Images')}\n"
"[/FILE]\n"
)
})
page_images = _pdf_to_page_images(p, max_pages=12, zoom=1.5)
handled_file = True
if not page_images:
raise RuntimeError(
get_ui_text(load_config(), "o_PDF_Convert_Failed")
)
for img_path in page_images:
blocks.append({
"type": "image_url",
"image_url": {
"url": _data_url_for_image(img_path)
}
})
elif p.suffix.lower() == ".docx":
file_text = _safe_read_docx(p)
file_type = "docx_text"
elif p.suffix.lower() == ".xlsx":
file_text = _safe_read_xlsx(p)
file_type = "xlsx_text"
if file_text is not None:
blocks.append({
"type": "text",
"text": (
"\n[FILE]\n"
f"name: {name}\n"
f"source_path_for_app_only: {str(p)}\n"
f"mime: {mime}\n"
f"editable: {'true' if editable else 'false'}\n"
f"type: {file_type}\n"
f"IMPORTANT: {get_ui_text(load_config(), 'o_File_Content_Already_Extracted')}\n"
"[CONTENT]\n"
f"{file_text}\n"
"[/CONTENT]\n"
"[/FILE]\n"
)
})
elif not handled_file:
blocks.append({
"type": "text",
"text": (
"\n[FILE]\n"
f"name: {name}\n"
f"path: {str(p)}\n"
f"mime: {mime}\n"
f"editable: {'true' if editable else 'false'}\n"
"type: unsupported\n"
f"note: {get_ui_text(load_config(), 'o_File_Note_Unsupported_Text_Read')}\n"
"[/FILE]\n"
)
})
if not blocks:
blocks = [{"type": "text", "text": ""}]
return blocks, (str(cached_image) if cached_image else None)
def fix_mojibake(s: str) -> str:
if not s:
return s
# Mojibake tipik işaretleri
if ("Ã" not in s) and ("Å" not in s) and ("â" not in s):
return s
try:
# yanlışlıkla latin-1/cp1252 gibi decode edilmiş UTF-8'i düzelt
return s.encode("latin-1").decode("utf-8")
except Exception:
return s
def _message_to_plain_text(content) -> str:
if isinstance(content, str):
return content.strip()
if isinstance(content, list):
parts = []
for item in content:
if not isinstance(item, dict):
continue
if item.get("type") == "text":
txt = item.get("text")
if isinstance(txt, str) and txt.strip():
parts.append(txt.strip())
return "\n".join(parts).strip()
return ""
def lang_code_to_name(code: str) -> str:
m = {
"tr": "Turkish",
"en": "English",
"de": "German",
"fr": "French",
"es": "Spanish",
}
return m.get(code, code)
def get_ui_language():
try:
if CONFIG_PATH.exists():
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
cfg = json.load(f) or {}
return str(cfg.get("ui_language", "en")).strip().lower()
except:
pass
return "en"
def _extract_images_from_message(msg: dict) -> dict:
"""
OpenRouter / farklı provider cevaplarından image url veya base64 çıkar.
UI'nin finalize_ai_response() fonksiyonunun anlayacağı formatı döndürür.
"""
out = {
"type": "text",
"content": "",
}
if not isinstance(msg, dict):
return out
content = msg.get("content")
text_parts = []
image_urls = []
image_b64 = []
# 1) content string ise
if isinstance(content, str):
if content.strip():
text_parts.append(content.strip())
# 2) content liste ise
elif isinstance(content, list):
for item in content:
if not isinstance(item, dict):
continue
itype = str(item.get("type") or "").strip().lower()
if itype == "text":
txt = item.get("text")
if isinstance(txt, str) and txt.strip():
text_parts.append(txt.strip())
elif itype in ("image_url", "output_image"):
iu = item.get("image_url")
if isinstance(iu, str) and iu.strip():
image_urls.append(iu.strip())
elif isinstance(iu, dict):
u = iu.get("url")
if isinstance(u, str) and u.strip():
image_urls.append(u.strip())
b64 = item.get("b64_json")
if isinstance(b64, str) and b64.strip():
image_b64.append(b64.strip())
data = item.get("image")
if isinstance(data, str) and data.strip():
image_b64.append(data.strip())
# 3) message level fallback alanlar
for key in ("image", "url"):
val = msg.get(key)
if isinstance(val, str) and val.strip():
image_urls.append(val.strip())
one_b64 = msg.get("image_base64")
if isinstance(one_b64, str) and one_b64.strip():
image_b64.append(one_b64.strip())
imgs = msg.get("images")
if isinstance(imgs, list):
for item in imgs:
if isinstance(item, str) and item.strip():
image_urls.append(item.strip())
elif isinstance(item, dict):
u = item.get("url")
if isinstance(u, str) and u.strip():
image_urls.append(u.strip())
iu = item.get("image_url")
if isinstance(iu, str) and iu.strip():
image_urls.append(iu.strip())
elif isinstance(iu, dict):
uu = iu.get("url")
if isinstance(uu, str) and uu.strip():
image_urls.append(uu.strip())
b64 = item.get("b64_json")
if isinstance(b64, str) and b64.strip():
image_b64.append(b64.strip())
# uniq
def _uniq(seq):
seen = set()
out2 = []
for x in seq:
if x not in seen:
seen.add(x)
out2.append(x)
return out2
image_urls = _uniq(image_urls)
image_b64 = _uniq(image_b64)
out["content"] = "\n".join(text_parts).strip()
if image_urls:
out["type"] = "image"
out["images"] = image_urls
if image_b64:
out["type"] = "image"
out["images_base64"] = image_b64
if image_urls and image_b64:
out["type"] = "image"
return out
def _extract_images_from_anywhere(obj) -> dict:
"""
OpenRouter / provider response içinde image url veya base64 farklı yerlerde dönebilir.
Bu fallback tüm response JSON içinde güvenli şekilde image benzeri alanları arar.
"""
out = {
"type": "text",
"content": "",
}
image_urls = []
image_b64 = []
image_key_names = {
"image",
"url",
"image_url",
"output_image",
"b64_json",
"image_base64",
"base64",
"data",
}
def add_url(s: str):
s = str(s or "").strip()
if not s:
return
if s.startswith("data:image/"):
image_urls.append(s)
return
low = s.lower()
if low.startswith("http://") or low.startswith("https://"):
if any(x in low for x in [".png", ".jpg", ".jpeg", ".webp", ".gif"]) or "image" in low:
image_urls.append(s)
def add_b64(s: str):
s = str(s or "").strip()
if not s:
return
if s.startswith("data:image/"):
image_urls.append(s)
return
# çok kısa stringleri base64 sanma
if len(s) < 500:
return
# basit base64 karakter kontrolü
if re.fullmatch(r"[A-Za-z0-9+/=\s]+", s):
image_b64.append(s)
def walk(x, parent_key=""):
if isinstance(x, dict):
for k, v in x.items():
key = str(k or "").strip().lower()
if isinstance(v, str):
if key in image_key_names or "image" in key or "b64" in key:
if key in ("b64_json", "image_base64", "base64") or "b64" in key:
add_b64(v)
else:
add_url(v)
add_b64(v)
else:
# data:image/... bazen key alakasız olabilir
if v.strip().startswith("data:image/"):
add_url(v)
else:
walk(v, key)
elif isinstance(x, list):
for item in x:
walk(item, parent_key)
elif isinstance(x, str):
if x.strip().startswith("data:image/"):
add_url(x)
walk(obj)
def uniq(seq):
seen = set()
result = []
for item in seq:
key = item[:200] + str(len(item))
if key in seen:
continue
seen.add(key)
result.append(item)
return result
image_urls = uniq(image_urls)
image_b64 = uniq(image_b64)
if image_urls:
out["type"] = "image"
out["images"] = image_urls
if image_b64:
out["type"] = "image"
out["images_base64"] = image_b64
return out
def _split_local_model(model_name: str) -> tuple[str, str]:
raw = str(model_name or "").strip()
if not raw or "/" not in raw:
cfg = load_config()
_die(get_ui_text(cfg, "o_Invalid_Local_Model_Format", model=raw), 1)
provider, actual_model = raw.split("/", 1)
provider = provider.strip().lower()
actual_model = actual_model.strip()
if not provider or not actual_model:
cfg = load_config()
_die(get_ui_text(cfg, "o_Invalid_Local_Model_Format", model=raw), 1)
return provider, actual_model
def _is_image_generation_model(model_name: str) -> bool:
m = str(model_name or "").strip().lower()
return any(x in m for x in [
"image",
"gpt-image",
"imagen",
"flux",
"recraft",
"stable-diffusion"
])
def _extract_image_settings_from_text(text: str) -> tuple[dict, str]:
raw = str(text or "")
m = re.search(
r"image\s*\{\s*(.*?)\s*\}",
raw,
flags=re.IGNORECASE | re.DOTALL
)
if not m:
return {}, raw
body = m.group(1)
settings = {}
for part in body.split(","):
part = part.strip()
if not part or ":" not in part:
continue
key, value = part.split(":", 1)
key = key.strip().lower()
value = value.strip()
if key and value:
settings[key] = value
clean_text = (raw[:m.start()] + raw[m.end():]).strip()
return settings, clean_text
def _extract_image_settings_from_final_messages(final_messages: list[dict]) -> dict:
found = {}
for msg in reversed(final_messages):
if not isinstance(msg, dict):
continue
if msg.get("role") != "user":
continue
content = msg.get("content")
if isinstance(content, str):
settings, clean = _extract_image_settings_from_text(content)
if settings:
msg["content"] = clean
found.update(settings)
return found
elif isinstance(content, list):
for item in content:
if not isinstance(item, dict):
continue
if item.get("type") != "text":
continue
txt = item.get("text")
if not isinstance(txt, str):
continue
settings, clean = _extract_image_settings_from_text(txt)
if settings:
item["text"] = clean
found.update(settings)
return found
return found
def _normalize_image_size_from_settings(settings: dict) -> str | None:
resolution = str(settings.get("resolution") or "").strip().lower()
aspect_ratio = str(settings.get("aspect_ratio") or "").strip().lower()
# Öncelik gerçek resolution
if resolution:
resolution = resolution.replace(" ", "")
if re.fullmatch(r"\d{3,5}x\d{3,5}", resolution):
return resolution
# Resolution yoksa aspect_ratio'dan yaygın size üret
ratio_map = {
"1:1": "1024x1024",
"16:9": "1920x1080",
"9:16": "1080x1920",
"4:3": "1600x1200",
"3:4": "1200x1600",
"3:2": "1536x1024",
"2:3": "1024x1536",
}
return ratio_map.get(aspect_ratio)
def _apply_image_settings_to_payload(payload: dict, settings: dict):
if not isinstance(payload, dict) or not isinstance(settings, dict):
return
size = _normalize_image_size_from_settings(settings)
if size:
payload["size"] = size
aspect_ratio = str(settings.get("aspect_ratio") or "").strip().lower()
if aspect_ratio:
payload["aspect_ratio"] = aspect_ratio
quality = str(settings.get("quality") or "").strip().lower()
if quality:
payload["quality"] = quality
def _build_image_settings_instruction(settings: dict, cfg: dict | None = None) -> str:
cfg = cfg or load_config()
if not isinstance(settings, dict) or not settings:
return ""
size = _normalize_image_size_from_settings(settings)
aspect_ratio = str(settings.get("aspect_ratio") or "").strip()
quality = str(settings.get("quality") or "").strip()
style = str(settings.get("style") or "").strip()
parts = []
if size:
parts.append(get_ui_text(cfg, "o_Image_Output_Size", size=size))
if aspect_ratio:
parts.append(get_ui_text(cfg, "o_Image_Output_Aspect_Ratio", aspect_ratio=aspect_ratio))
if quality:
parts.append(get_ui_text(cfg, "o_Image_Output_Quality", quality=quality))
if style:
parts.append(get_ui_text(cfg, "o_Image_Output_Style", style=style))
if not parts:
return ""
return (
get_ui_text(cfg, "o_Image_Output_Preferences_Title")
+ "\n"
+ "\n".join(f"- {p}" for p in parts)
+ "\n"
+ get_ui_text(cfg, "o_Image_Output_Preferences_Note")
)
def _append_image_instruction_to_last_user(final_messages: list[dict], instruction: str):
instruction = str(instruction or "").strip()
if not instruction:
return
for msg in reversed(final_messages):
if not isinstance(msg, dict):
continue
if msg.get("role") != "user":
continue
content = msg.get("content")
if isinstance(content, list):
content.append({
"type": "text",
"text": "\n\n" + instruction
})
return
if isinstance(content, str):
msg["content"] = content.rstrip() + "\n\n" + instruction
return
def _normalize_model_entry(item) -> dict | None:
if isinstance(item, dict):
mid = str(item.get("id") or "").strip()
if not mid:
return None
return {
"id": mid,
"local": bool(item.get("local", False))
}
if isinstance(item, str):
mid = item.strip()
if not mid:
return None
return {
"id": mid,
"local": False
}
return None
def _get_models_list(cfg: dict) -> list[dict]:
raw = cfg.get("ai_models", [])
if not isinstance(raw, list):
raw = []
out = []
seen = set()
for item in raw:
norm = _normalize_model_entry(item)
if not norm:
continue
mid = norm["id"]
if mid in seen:
continue
seen.add(mid)
out.append(norm)
return out
def _get_chat_model_entry(cfg: dict, chat_file: Path) -> dict:
models = _get_models_list(cfg)
if not models:
_die(get_ui_text(cfg, "o_AI_Models_Empty"), 1)
model_ids = {m["id"] for m in models}
chat_models = cfg.get("chat_models", {})
if not isinstance(chat_models, dict):
chat_models = {}
chat_key = chat_file.name
selected = chat_models.get(chat_key)
chosen_id = ""
chosen_local = False
if isinstance(selected, dict):
chosen_id = str(selected.get("id") or "").strip()
chosen_local = bool(selected.get("local", False))
elif isinstance(selected, str):
chosen_id = selected.strip()
chosen_local = False
if not chosen_id or chosen_id not in model_ids:
fallback = models[0]
chosen_id = fallback["id"]
chosen_local = bool(fallback.get("local", False))
chat_models[chat_key] = {
"id": chosen_id,
"local": chosen_local
}
cfg["chat_models"] = chat_models
_save_config(cfg)
return {
"id": chosen_id,
"local": chosen_local
}
# ai_models içindeki gerçek local değerini koru
for m in models:
if m["id"] == chosen_id:
chosen_local = bool(m.get("local", False))
break
fixed_entry = {
"id": chosen_id,
"local": chosen_local
}
if selected != fixed_entry:
chat_models[chat_key] = fixed_entry
cfg["chat_models"] = chat_models
_save_config(cfg)
return fixed_entry
def _pick_model(cfg: dict, chat_file: Path, cli_model: str | None) -> dict:
if cli_model:
return {
"id": str(cli_model).strip(),
"local": False
}
return _get_chat_model_entry(cfg, chat_file)
def _coerce_number(val: str):
s = str(val or "").strip()
if not s:
return None
try:
if "." in s:
return float(s)
return int(s)
except Exception:
return None
def _build_local_options(provider_cfg: dict) -> dict:
opts = {}
num = _coerce_number(provider_cfg.get("temperature", ""))
if num is not None:
opts["temperature"] = num
num = _coerce_number(provider_cfg.get("top_p", ""))
if num is not None:
opts["top_p"] = num
num = _coerce_number(provider_cfg.get("top_k", ""))
if num is not None:
opts["top_k"] = num
num = _coerce_number(provider_cfg.get("repeat_penalty", ""))
if num is not None:
opts["repeat_penalty"] = num
num = _coerce_number(provider_cfg.get("num_ctx", ""))
if num is not None:
opts["num_ctx"] = num
num = _coerce_number(provider_cfg.get("num_predict", ""))
if num is not None:
opts["num_predict"] = num
keep_alive = str(provider_cfg.get("keep_alive") or "").strip()
if keep_alive:
opts["keep_alive"] = keep_alive
return opts
def _flatten_messages_for_local(final_messages: list[dict]) -> str:
parts = []
for msg in final_messages:
if not isinstance(msg, dict):
continue
role = str(msg.get("role") or "").strip().upper()
content = msg.get("content")
if isinstance(content, str):
txt = content.strip()
if txt:
parts.append(f"[{role}]\n{txt}")
continue
if isinstance(content, list):
subparts = []
for item in content:
if not isinstance(item, dict):
continue
itype = str(item.get("type") or "").strip().lower()
if itype == "text":
txt = str(item.get("text") or "").strip()
if txt:
subparts.append(txt)
elif itype == "image_url":
iu = item.get("image_url")
if isinstance(iu, dict):
u = str(iu.get("url") or "").strip()
else:
u = str(iu or "").strip()
if u:
subparts.append(f"[IMAGE]\n{u}")
joined = "\n".join(x for x in subparts if x.strip()).strip()
if joined:
parts.append(f"[{role}]\n{joined}")
return "\n\n".join(parts).strip()
def _call_ollama(
provider_cfg: dict,
actual_model: str,
final_messages: list[dict],
image_settings: dict | None = None,
is_image_model: bool = False
) -> str:
base_url = str(provider_cfg.get("base_url") or "").strip() or "http://127.0.0.1:11434"
url = base_url.rstrip("/") + "/api/generate"
prompt = _flatten_messages_for_local(final_messages)
payload = {
"model": actual_model,
"prompt": prompt,
"stream": False,
}
if image_settings and is_image_model:
_apply_image_settings_to_payload(payload, image_settings)
system_prompt = str(provider_cfg.get("system_prompt") or "").strip()
if system_prompt:
payload["system"] = system_prompt
options = _build_local_options(provider_cfg)
if options:
payload["options"] = options
r = requests.post(
url,
json=payload,
timeout=300
)
if r.status_code != 200:
cfg = load_config()
_die(get_ui_text(cfg, "o_Ollama_API_Error", status_code=r.status_code, details=r.text[:1000]), 1)
try:
data = r.json()
except Exception as e:
cfg = load_config()
_die(get_ui_text(cfg, "o_Ollama_JSON_Parse_Error", error=e), 1)
text = str(data.get("response") or "").strip()
if not text:
cfg = load_config()
_die(get_ui_text(cfg, "o_Ollama_Empty_Response"), 1)
result_obj = {
"type": "text",
"content": fix_mojibake(text)
}
prompt_eval = data.get("prompt_eval_count")
eval_count = data.get("eval_count")
if prompt_eval is not None or eval_count is not None:
p = int(prompt_eval or 0)
c = int(eval_count or 0)
result_obj["usage"] = {
"prompt_tokens": p,
"completion_tokens": c,
"total_tokens": p + c
}
return json.dumps(result_obj, ensure_ascii=False)
def main():
try:
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
if hasattr(sys.stderr, "reconfigure"):
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# Ekstra sağlamlaştırma (bazı ortamlarda fark yaratır)
os.environ["PYTHONIOENCODING"] = "utf-8"
# Args
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument("chat_file", nargs="?")
parser.add_argument("selected_indexes", nargs="?")
parser.add_argument("--model", dest="model", default=None)
parser.add_argument("--context-mode", dest="context_mode", default="rag")
parser.add_argument("--approved-web-search-query", dest="approved_web_search_query", default=None)
if len(sys.argv) < 2:
sys.exit(1)
args, _unknown = parser.parse_known_args()
context_mode = str(args.context_mode or "rag").strip().lower()
if context_mode not in ("rag", "direct"):
context_mode = "rag"
if not args.chat_file:
sys.exit(1)
chat_file = Path(args.chat_file)
if not chat_file.exists():
sys.exit(1)
selected_indexes = []
if args.selected_indexes:
try:
selected_indexes = _parse_selected_indexes(args.selected_indexes)
except Exception:
selected_indexes = []
# Config
cfg = _load_config()
response_style = (cfg.get("response_style") or "").strip()
# model seçimi (chat’e göre)
model_entry = _pick_model(cfg, chat_file, args.model)
model_name = str(model_entry["id"]).strip()
is_local_model = bool(model_entry.get("local", False))
local_providers = cfg.get("local_providers", {})
if not isinstance(local_providers, dict):
local_providers = {}
API_KEY = ""
if not is_local_model:
API_KEY = (cfg.get("open_router_key") or "").strip()
if not API_KEY:
_die(get_ui_text(cfg, "o_OpenRouter_Key_Missing"), 1)
# read rag settings
rag_settings = cfg.get("rag_settings", {})
if not isinstance(rag_settings, dict):
rag_settings = {}
def rag_bool(key, default):
return bool(rag_settings.get(key, default))
def rag_int(key, default, min_v=0, max_v=100_000):
try:
val = int(rag_settings.get(key, default))
except Exception:
val = default
return max(min_v, min(max_v, val))
ask_for_web_search = bool(cfg.get("ask_for_web_search", True))
approved_web_search_query = str(args.approved_web_search_query or "").strip()
recent_message_count = rag_int("recent_message_count", 10, 0, 50)
retrieved_chunk_count = rag_int("retrieved_chunk_count", 5, 0, 50)
summary_update_every = rag_int("summary_update_every", 20, 1, 200)
memory_chunk_max_chars = rag_int("memory_chunk_max_chars", 1200, 100, 100_000)
summary_max_chars = rag_int("summary_max_chars", 6000, 100, 100_000)
code_context_max_chars = rag_int("code_context_max_chars", 8000, 100, 100_000)
use_summary = rag_bool("use_summary", True)
use_recent_messages = rag_bool("use_recent_messages", True)
use_retrieval = rag_bool("use_retrieval", True)
use_code_context = rag_bool("use_code_context", True)
include_recent_attachments = rag_bool("include_recent_attachments", False)
# Read chat
try:
chat_data = load_chat_data(chat_file)
messages = get_chat_messages(chat_data)
except Exception as e:
_die(get_ui_text(cfg, "o_Chat_Read_Error", error=e), 1)
if not messages:
sys.exit(0)
if messages and isinstance(messages[-1], dict) and messages[-1].get("streaming") and messages[-1].get("role") in ("bot", "assistant"):
messages = messages[:-1] # placeholder'ı payload'a sokma
if not messages:
sys.exit(0)
# --- son gerçek user mesajını bul ---
last_user = None
for m in reversed(messages):
if isinstance(m, dict) and m.get("role") == "user":
last_user = m
break
if last_user is None:
_die(get_ui_text(cfg, "o_No_User_Message"), 1)
last = last_user
final_messages = []
editable_map = {} # path(str) -> bool
# Son user mesajını memory sistemine ekle
last_text = str(last.get("content") or "").strip()
files_last = last.get("files") or []
is_pdf_text_only_request = False
is_pdf_image_only_request = False
is_pdf_mixed_request = False
is_pdf_mixed_image_only_request = False
mixed_pdf_layout = None
mixed_pdf_source = None
if isinstance(files_last, list):
for ff in files_last:
if not isinstance(ff, dict):
continue
p = Path(str(ff.get("path") or ""))
editable = bool(ff.get("edit"))
if editable and p.suffix.lower() == ".pdf":
analysis = _analyze_pdf_kind(p)
kind = analysis.get("kind")
if kind == "text_only":
is_pdf_text_only_request = True
elif kind == "mixed":
mixed_pdf_source = p
mixed_pdf_layout = _extract_pdf_mixed_layout(p)
if mixed_pdf_layout and mixed_pdf_layout.get("pages"):
is_pdf_mixed_request = True
else:
# Layout çıkarılamıyorsa güvenli fallback:
# metin varsa text-only gibi davran.
fallback_text = analysis.get("text") or _safe_read_pdf(p)
if len((fallback_text or "").strip()) >= 80:
is_pdf_text_only_request = True
else:
is_pdf_image_only_request = True
else:
is_pdf_image_only_request = True
break
if context_mode == "rag":
# Son user mesajını memory sistemine ekle
if last_text:
add_memory_chunk(
chat_data,
last_text[:memory_chunk_max_chars],
source="chat:user"
)
# Eğer son mesajda kod dosyası varsa code_context oluştur
if use_code_context:
update_code_context_from_message(chat_data, last)
selected_prompt_blocks = cfg.get("prompt_chooser_blocks", ["copyable"])
selected_prompt_blocks_set = {
str(x or "").strip()
for x in selected_prompt_blocks
if str(x or "").strip()
}
pdf_mode_text = "pdf_text" in selected_prompt_blocks_set
pdf_mode_image = "pdf_image" in selected_prompt_blocks_set
pdf_mode_text_image = "pdf_text_image" in selected_prompt_blocks_set
if is_pdf_mixed_request:
# Mixed PDF kararını keyword tahminiyle vermiyoruz.
# Kullanıcının açık seçimi Prompt Chooser üzerinden gelir:
# - pdf_text -> sadece metin blokları
# - pdf_image -> sadece görsel blokları / sayfa görseli
# - pdf_text_image -> metin + görsel layout birlikte
#
# Bu yüzden _user_wants_image_edit() / _user_wants_text_edit()
# burada kullanılmaz.
is_pdf_mixed_image_only_request = bool(pdf_mode_image and not pdf_mode_text_image and not pdf_mode_text)
has_editable_pdf_request = (
is_pdf_text_only_request
or is_pdf_image_only_request
or is_pdf_mixed_request
)
if has_editable_pdf_request and not (pdf_mode_text or pdf_mode_image or pdf_mode_text_image):
msg = get_ui_text(cfg, "o_PDF_Mode_Required")
result_obj = {
"type": "text",
"content": msg
}
sys.stdout.write(json.dumps(result_obj, ensure_ascii=False))
sys.stdout.flush()
sys.exit(0)
system_prompt = build_global_system_prompt(selected_prompt_blocks, cfg)
if is_pdf_text_only_request and pdf_mode_text:
system_prompt += "\n\n" + get_prompt_text(cfg, PDF_RULE_KEYS["pdf_text_only_to_docx"])
elif is_pdf_image_only_request and pdf_mode_image:
system_prompt += "\n\n" + get_prompt_text(cfg, PDF_RULE_KEYS["pdf_image"])
elif is_pdf_mixed_request:
if pdf_mode_text:
system_prompt += "\n\n" + get_prompt_text(cfg, PDF_RULE_KEYS["pdf_mixed_text_only"])
elif pdf_mode_image:
system_prompt += "\n\n" + get_prompt_text(cfg, PDF_RULE_KEYS["pdf_image"])
elif pdf_mode_text_image:
system_prompt += "\n\n" + get_prompt_text(cfg, PDF_RULE_KEYS["pdf_mixed_separate"])
final_messages.append({
"role": "system",
"content": system_prompt
})
if context_mode == "rag" and (use_summary or use_retrieval or use_code_context):
memory_context = build_memory_context(chat_data, last_text)
if not use_summary:
# Basit filtre: summary başlığı varsa kaldırmak için memory.py tarafında daha temiz yapılabilir.
pass
if memory_context:
memory_context = memory_context[:summary_max_chars + code_context_max_chars + 4000]
final_messages.append({
"role": "system",
"content": (
get_ui_text(
cfg,
"o_RAG_Context_System_Message"
) + "\n\n" + memory_context
)
})
cfg = load_config()
force_lang = bool(cfg.get("force_ui_language", False))
lang = get_ui_language()
lang_name = lang_code_to_name(lang)
if force_lang:
final_messages.append({
"role": "system",
"content": get_ui_text(cfg, "o_System_Force_Language", lang_name=lang_name)
})
if response_style:
final_messages.append({
"role": "system",
"content": get_ui_text(cfg, "o_System_Response_Style", response_style=response_style)
})
# ---------------- RECENT MESSAGES MEMORY ----------------
# Burada sadece önceki son 10 konuşma mesajını ekliyoruz.
if context_mode == "rag" and use_recent_messages and recent_message_count > 0:
recent_messages = []
lookback_count = recent_message_count + 2
for m in messages[-lookback_count:]:
if not isinstance(m, dict):
continue
if m.get("streaming"):
continue
# Son user mesajını tekrar ekleme
if m is last:
continue
role_raw = m.get("role")
if role_raw not in ("user", "bot", "assistant"):
continue
recent_messages.append(m)
recent_messages = recent_messages[-recent_message_count:]
if recent_messages:
final_messages.append({
"role": "system",
"content": get_ui_text(
cfg,
"o_RAG_Recent_Messages_System_Message"
)
})
for m in recent_messages:
role = _role_map(m.get("role"))
has_any_attach = False
if m.get("image") and Path(str(m["image"])).exists():
has_any_attach = True
imgs = m.get("images")
if isinstance(imgs, list) and any(Path(str(p)).exists() for p in imgs):
has_any_attach = True
files = m.get("files")
if isinstance(files, list) and any(
Path(str(f.get("path") or "")).exists()
for f in files
if isinstance(f, dict)
):
has_any_attach = True
if has_any_attach and include_recent_attachments:
blocks, _ = _build_blocks_and_cache_info(m, cache_images_dir=None)
final_messages.append({
"role": role,
"content": blocks
})
else:
final_messages.append({
"role": role,
"content": _message_content_with_generated_files(m)
})
# Cache setup
REFS_CACHE_DIR.mkdir(parents=True, exist_ok=True)
# Reference system (TEXT + IMAGE) + CACHE
if selected_indexes:
chat_key = _safe_chat_key(chat_file)
bundle_id = _utc_stamp()
bundle_dir = REFS_CACHE_DIR / chat_key / bundle_id
bundle_images_dir = bundle_dir / "images"
bundle_dir.mkdir(parents=True, exist_ok=True)
final_messages.append({
"role": "system",
"content": get_ui_text(cfg, "o_System_Reference_Messages")
})
cache_items = []
for i in selected_indexes:
if i < len(messages) - 1:
msg = messages[i]
role = _role_map(msg.get("role"))
has_any_attach = False
# editable_map doldur (referans mesajın file edit izinleri)
files_meta = msg.get("files") or []
if isinstance(files_meta, list):
for ff in files_meta:
if not isinstance(ff, dict):
continue
p_raw = str(ff.get("path") or "").strip()
if not p_raw:
continue
editable_map[p_raw] = bool(ff.get("edit"))
# images?
if msg.get("image") and Path(str(msg["image"])).exists():
has_any_attach = True
imgs = msg.get("images")
if isinstance(imgs, list) and any(Path(str(p)).exists() for p in imgs):
has_any_attach = True
# files?
files = msg.get("files")
if isinstance(files, list) and any(Path(str(f.get("path") or "")).exists() for f in files if isinstance(f, dict)):
has_any_attach = True
if has_any_attach:
blocks, cached_img = _build_blocks_and_cache_info(msg, bundle_images_dir)
final_messages.append({"role": role, "content": blocks})
else:
final_messages.append({"role": role, "content": msg.get("content", "")})
cached_img = None
cache_items.append({
"index": i,
"role": msg.get("role", ""),
"content": msg.get("content", ""),
"image_original": msg.get("image", None),
"images_original": msg.get("images", None),
"files_original": msg.get("files", None),
"image_cached": cached_img
})
bundle = {
"bundle_id": bundle_id,
"created_at_utc": datetime.now(timezone.utc).isoformat(),
"chat_file": str(chat_file),
"selected_indexes": selected_indexes,
"items": cache_items
}
try:
(bundle_dir / "bundle.json").write_text(
json.dumps(bundle, ensure_ascii=False, indent=2),
encoding="utf-8"
)
except Exception:
pass
# Last message (answer) - TEXT + (image / images)
# last message editable_map doldur
if isinstance(files_last, list):
for ff in files_last:
if not isinstance(ff, dict):
continue
p_raw = str(ff.get("path") or "").strip()
if not p_raw:
continue
editable_map[p_raw] = bool(ff.get("edit"))
# Last message (answer) - TEXT + (images/files)
role = _role_map(last.get("role"))
has_any_attach = False
# images?
if last.get("image") and Path(str(last["image"])).exists():
has_any_attach = True
imgs = last.get("images")
if isinstance(imgs, list) and any(Path(str(p)).exists() for p in imgs):
has_any_attach = True
# files?
files = last.get("files")
if isinstance(files, list) and any(
Path(str(f.get("path") or "")).exists()
for f in files
if isinstance(f, dict)
):
has_any_attach = True
if has_any_attach:
pdf_mode_for_blocks = "auto"
if pdf_mode_text:
pdf_mode_for_blocks = "text"
elif pdf_mode_image:
pdf_mode_for_blocks = "image"
elif pdf_mode_text_image:
pdf_mode_for_blocks = "auto"
blocks, _ = _build_blocks_and_cache_info(
last,
cache_images_dir=None,
pdf_mode=pdf_mode_for_blocks
)
final_messages.append({"role": role, "content": blocks})
else:
final_messages.append({"role": role, "content": last.get("content", "")})
# Payload
reply = ""
usage_obj = None
is_image_model = _is_image_generation_model(model_name)
image_settings = _extract_image_settings_from_final_messages(final_messages)
image_instruction = _build_image_settings_instruction(image_settings, cfg)
if image_instruction and is_image_model:
_append_image_instruction_to_last_user(
final_messages,
image_instruction
)
try:
if is_local_model:
provider, actual_model = _split_local_model(model_name)
provider_cfg = local_providers.get(provider)
if not isinstance(provider_cfg, dict):
_die(get_ui_text(cfg, "o_Local_Provider_Not_Found", provider=provider), 1)
if not bool(provider_cfg.get("enabled", False)):
_die(get_ui_text(cfg, "o_Local_Provider_Disabled", provider=provider), 1)
run_startup = str(provider_cfg.get("run_startup") or "").strip()
if run_startup:
try:
subprocess.Popen(
run_startup,
shell=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
except Exception:
pass
else:
# chat'e mesaj döndür (stdout)
result_obj = {
"type": "text",
"content": get_ui_text(cfg, "o_Start_Local_Server_From_App", provider=provider)
}
sys.stdout.write(json.dumps(result_obj, ensure_ascii=False))
sys.stdout.flush()
sys.exit(1)
if provider == "ollama":
reply = _call_ollama(
provider_cfg,
actual_model,
final_messages,
image_settings=image_settings,
is_image_model=is_image_model
)
parsed_reply = json.loads(reply)
assistant_text = str(parsed_reply.get("content") or "").strip()
web_query = _extract_web_search_call(assistant_text)
if web_query and ("web_search" in selected_prompt_blocks_set):
web_results = _tavily_search(cfg, web_query, max_results=5)
final_messages.append({
"role": "assistant",
"content": assistant_text
})
final_messages.append({
"role": "system",
"content": (
get_ui_text(cfg, "o_System_Web_Search_Results")
+ "\n\n"
+ web_results
)
})
reply = _call_ollama(
provider_cfg,
actual_model,
final_messages,
image_settings=image_settings,
is_image_model=is_image_model
)
sys.stdout.write(reply)
sys.stdout.flush()
try:
parsed_reply = json.loads(reply)
assistant_text = str(parsed_reply.get("content") or "").strip() if isinstance(parsed_reply, dict) else ""
if context_mode == "rag":
if assistant_text:
add_memory_chunk(
chat_data,
assistant_text[:memory_chunk_max_chars],
source="chat:assistant"
)
if use_summary and should_update_summary(chat_data, every=summary_update_every):
chat_data["summary"] = make_simple_summary(chat_data)[:summary_max_chars]
save_chat_data(chat_file, chat_data)
except Exception:
pass
sys.exit(0)
else:
_die(get_ui_text(cfg, "o_Unsupported_Local_Provider", provider=provider), 1)
# Remote / OpenRouter
if (is_pdf_text_only_request or is_pdf_mixed_request) and is_image_model:
result_obj = {
"type": "text",
"content": get_ui_text(cfg, "o_PDF_Has_Text_Select_Text_Model")
}
sys.stdout.write(json.dumps(result_obj, ensure_ascii=False))
sys.stdout.flush()
sys.exit(0)
# Text tabanlı PDF düzenleme isteklerinde image model kullanma.
# Çünkü image model metni düzenlemek yerine görsel üretmeye çalışır.
try:
for ff in files_last:
if not isinstance(ff, dict):
continue
p = Path(str(ff.get("path") or ""))
if p.suffix.lower() == ".pdf" and bool(ff.get("edit")):
has_editable_pdf = True
break
except Exception:
pass
selected_prompt_blocks = cfg.get("prompt_chooser_blocks", ["copyable"])
if not isinstance(selected_prompt_blocks, list):
selected_prompt_blocks = ["copyable"]
selected_prompt_blocks = {
str(x or "").strip()
for x in selected_prompt_blocks
}
has_editable_generated_doc = False
try:
for ff in files_last:
if not isinstance(ff, dict):
continue
p = Path(str(ff.get("path") or ""))
if p.suffix.lower() in (".docx", ".xlsx", ".txt", ".md", ".pdf") and bool(ff.get("edit")):
has_editable_generated_doc = True
break
except Exception:
pass
wants_file_create = (
has_editable_generated_doc
and ("file_create" in selected_prompt_blocks)
)
has_input_images = bool(
last.get("image")
or last.get("images")
or any(
isinstance(f, dict) and str(f.get("path") or "").lower().endswith((".png", ".jpg", ".jpeg", ".webp"))
for f in files_last
)
)
has_ref_images = False
try:
for i in selected_indexes:
if not (0 <= i < len(messages)):
continue
rm = messages[i]
if rm.get("image"):
has_ref_images = True
break
imgs = rm.get("images")
if isinstance(imgs, list) and imgs:
has_ref_images = True
break
files = rm.get("files")
if isinstance(files, list):
for f in files:
if not isinstance(f, dict):
continue
p = str(f.get("path") or "").lower()
if p.endswith((".png", ".jpg", ".jpeg", ".webp")):
has_ref_images = True
break
if has_ref_images:
break
except Exception:
has_ref_images = False
should_stream = (
(not is_image_model)
and (not has_input_images)
and (not has_ref_images)
and (not wants_file_create)
and (not is_pdf_text_only_request)
and (not is_pdf_image_only_request)
and (not is_pdf_mixed_request)
)
web_search_enabled = "web_search" in selected_prompt_blocks
if web_search_enabled:
should_stream = False
payload = {
"model": model_name,
"messages": final_messages,
"max_tokens": 5120,
"stream": should_stream
}
if image_settings and is_image_model:
_apply_image_settings_to_payload(payload, image_settings)
response = requests.post(
OPENROUTER_URL,
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
"HTTP-Referer": "http://localhost",
"X-Title": "capture-ai"
},
json=payload,
timeout=120,
stream=should_stream
)
if response.status_code != 200:
sys.stderr.write(get_ui_text(cfg, "o_OpenRouter_API_Error", status_code=response.status_code, details=response.text) + "\n")
sys.exit(1)
if not should_stream:
data = response.json()
usage_data = data.get("usage")
if usage_data:
prompt_t = int(usage_data.get("prompt_tokens", 0) or 0)
comp_t = int(usage_data.get("completion_tokens", 0) or 0)
total_t = int(usage_data.get("total_tokens", 0) or 0)
usage_obj = {
"prompt_tokens": prompt_t,
"completion_tokens": comp_t,
"total_tokens": total_t
}
msg = {}
try:
msg = data["choices"][0]["message"]
except Exception:
msg = {}
result_obj = _extract_images_from_message(msg)
# Bazı image modelleri görseli message içinde değil,
# response JSON'un başka alanlarında döndürebiliyor.
# Ama fallback image bulamazsa mevcut text content'i ezmemeli.
if not result_obj.get("images") and not result_obj.get("images_base64"):
fallback_images = _extract_images_from_anywhere(data)
if fallback_images.get("images") or fallback_images.get("images_base64"):
old_content = str(result_obj.get("content") or "").strip()
result_obj = fallback_images
if old_content and not str(result_obj.get("content") or "").strip():
result_obj["content"] = old_content
# AI bazen dosya metadata'sını message.content içinde JSON string olarak döndürür.
# Onu gerçek result_obj haline çeviriyoruz.
content_text = str(result_obj.get("content") or "").strip()
if content_text.startswith("{"):
try:
parsed_content = json.loads(content_text)
if isinstance(parsed_content, dict) and (
parsed_content.get("generated_files_base64")
or parsed_content.get("generated_files_url")
or parsed_content.get("generated_files")
):
result_obj = parsed_content
except Exception:
pass
web_query = _extract_web_search_call(content_text)
if web_query and web_search_enabled:
if ask_for_web_search and not approved_web_search_query:
result_obj = {
"type": "text",
"content": "",
"web_search_request": {
"query": web_query,
"status": "pending"
}
}
print(json.dumps(result_obj, ensure_ascii=False))
return
query_to_search = approved_web_search_query or web_query
web_results = _tavily_search(cfg, query_to_search, max_results=5)
final_messages.append({
"role": "assistant",
"content": content_text
})
final_messages.append({
"role": "system",
"content": (
get_ui_text(cfg, "o_System_Web_Search_Results_With_Fallback")
+ "\n\n"
+ web_results
)
})
payload2 = {
"model": model_name,
"messages": final_messages,
"max_tokens": 5120,
"stream": False
}
if image_settings and is_image_model:
_apply_image_settings_to_payload(payload2, image_settings)
response2 = requests.post(
OPENROUTER_URL,
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
"HTTP-Referer": "http://localhost",
"X-Title": "capture-ai"
},
json=payload2,
timeout=120
)
if response2.status_code != 200:
sys.stderr.write(
get_ui_text(
cfg,
"o_OpenRouter_API_Error",
status_code=response2.status_code,
details=response2.text
) + "\n"
)
sys.exit(1)
data = response2.json()
msg = data["choices"][0]["message"]
result_obj = _extract_images_from_message(msg)
content_text = str(result_obj.get("content") or "").strip()
# Editable PDF isteğinde AI PDF değil image döndürürse,
# app image'ı PDF dosyasına paketler.
generated_files = []
# 1) PDF text -> AI file_create DOCX verir, app DOCX + PDF üretir
if is_pdf_text_only_request and pdf_mode_text:
reply_text = str(result_obj.get("content") or "").strip()
generated_files = _apply_file_create_from_reply(reply_text)
clean_text = _remove_named_block(reply_text, "file_create")
result_obj = {
"type": "text",
"content": clean_text or get_ui_text(cfg, "o_PDF_Text_Edited"),
"generated_files": generated_files,
}
# 2) PDF image + scan PDF -> AI PNG verir, app PDF'e çevirir
elif is_pdf_image_only_request and pdf_mode_image and (
result_obj.get("images") or result_obj.get("images_base64")
):
image_paths = _save_result_images_to_files(result_obj)
pdf_out = _image_paths_to_pdf_file(
image_paths,
output_name="edited-scan-pdf.pdf"
)
if pdf_out and pdf_out.exists():
generated_files.append({
"path": str(pdf_out.resolve()),
"name": pdf_out.name
})
result_obj = {
"type": "text",
"content": get_ui_text(cfg, "o_PDF_Image_Edited"),
"generated_files": generated_files,
}
# 3) Mixed PDF + sadece image -> AI PNG verir, app eski PDF üstüne image block olarak koyar
elif is_pdf_mixed_request and pdf_mode_image and (
result_obj.get("images") or result_obj.get("images_base64")
):
image_paths = _save_result_images_to_files(result_obj)
pdf_out = None
if mixed_pdf_source and mixed_pdf_layout:
pdf_out = _overlay_images_on_pdf_blocks(
mixed_pdf_source,
mixed_pdf_layout,
image_paths,
output_name="edited-pdf-image.pdf"
)
if pdf_out and pdf_out.exists():
generated_files.append({
"path": str(pdf_out.resolve()),
"name": pdf_out.name
})
result_obj = {
"type": "text",
"content": get_ui_text(cfg, "o_PDF_Image_Block_Edited"),
"generated_files": generated_files,
}
# 4) Mixed PDF + text only / text image -> AI JSON verir, app layout'u yeniden kurar
elif is_pdf_mixed_request and (pdf_mode_text or pdf_mode_text_image):
parsed = None
content_text = str(result_obj.get("content") or "").strip()
if content_text.startswith("{"):
try:
parsed = json.loads(content_text)
except Exception:
parsed = None
if isinstance(result_obj, dict) and result_obj.get("type") == "pdf_mixed_rewrite":
parsed = result_obj
if isinstance(parsed, dict) and parsed.get("type") == "pdf_mixed_rewrite":
text_replacements = parsed.get("text_replacements") or {}
image_replacements = parsed.get("image_replacements") or {}
pdf_out = _create_pdf_from_mixed_layout(
mixed_pdf_layout or {},
text_replacements,
image_replacements,
output_name="edited-mixed-pdf.pdf"
)
if pdf_out and pdf_out.exists():
generated_files.append({
"path": str(pdf_out.resolve()),
"name": pdf_out.name
})
result_obj = {
"type": "text",
"content": str(parsed.get("content") or get_ui_text(cfg, "o_PDF_Mixed_Edited")),
"generated_files": generated_files,
}
if is_pdf_mixed_request and isinstance(result_obj, dict):
content_text = str(result_obj.get("content") or "").strip()
if content_text.startswith("{"):
try:
parsed_content = json.loads(content_text)
if isinstance(parsed_content, dict):
result_obj = parsed_content
except Exception:
pass
if result_obj.get("type") == "pdf_mixed_rewrite":
text_replacements = result_obj.get("text_replacements")
image_replacements = result_obj.get("image_replacements")
# Kullanıcı sadece resim değişikliği istediyse,
# AI yazı üretse bile hepsini yok say.
if is_pdf_mixed_image_only_request:
text_replacements = {}
if mixed_pdf_layout and isinstance(text_replacements, dict):
out = _create_pdf_from_mixed_layout(
mixed_pdf_layout,
text_replacements,
image_replacements if isinstance(image_replacements, dict) else {},
"edited-mixed-pdf.pdf"
)
if out and out.exists():
result_obj = {
"type": "text",
"content": get_ui_text(cfg, "o_File_Created"),
"generated_files": [{
"path": str(out.resolve()),
"name": out.name
}]
}
if usage_obj is not None:
result_obj["usage"] = usage_obj
generated_files = []
# PDF girdiyse ve AI hâlâ PNG/image döndürdüyse:
# PNG kullanıcıya gösterilmesin, PDF'e çevrilsin.
if has_editable_pdf_request and (
result_obj.get("images")
or result_obj.get("images_base64")
or result_obj.get("image")
or result_obj.get("image_base64")
or result_obj.get("url")
):
image_paths = _save_result_images_to_files(result_obj)
pdf_out = None
if image_paths:
if is_pdf_mixed_request and mixed_pdf_source and mixed_pdf_layout:
pdf_out = _overlay_images_on_pdf_blocks(
mixed_pdf_source,
mixed_pdf_layout,
image_paths,
output_name="edited-pdf-image.pdf"
)
if not pdf_out:
pdf_out = _image_paths_to_pdf_file(
image_paths,
output_name="edited-pdf-result.pdf"
)
if pdf_out and pdf_out.exists():
generated_files.append({
"path": str(pdf_out.resolve()),
"name": pdf_out.name
})
result_obj = {
"type": "text",
"content": get_ui_text(cfg, "o_File_Created"),
"generated_files": generated_files,
}
elif isinstance(result_obj.get("generated_files"), list):
generated_files = result_obj["generated_files"]
else:
generated_files += _apply_ai_returned_files_from_result(result_obj)
generated_files += _apply_file_create_from_reply(result_obj.get("content", ""))
if generated_files:
clean_content = _remove_named_block(
result_obj.get("content", ""),
"file_create"
)
final_text = clean_content.strip()
if not final_text:
final_text = get_ui_text(cfg, "o_File_Created")
result_obj = {
"type": "text",
"content": final_text,
"generated_files": generated_files
}
if usage_obj is not None:
result_obj["usage"] = usage_obj
if not result_obj.get("content") and (
result_obj.get("images") or result_obj.get("images_base64")
):
result_obj["content"] = get_ui_text(cfg, "o_Image_Created")
# hiç image yoksa düz text fallback
if not result_obj.get("content") and not result_obj.get("images") and not result_obj.get("images_base64"):
txt = _message_to_plain_text(msg.get("content"))
result_obj = {
"type": "text",
"content": fix_mojibake(txt)
}
if usage_obj is not None:
result_obj["usage"] = usage_obj
reply = json.dumps(result_obj, ensure_ascii=False)
sys.stdout.write(reply)
sys.stdout.flush()
else:
reply_parts = []
usage_data = None
for raw_line in response.iter_lines(decode_unicode=False, chunk_size=1024):
if not raw_line:
continue
line = raw_line.strip()
if not line.startswith(b"data:"):
continue
data_str = line[len(b"data:"):].strip()
if data_str == b"[DONE]":
break
try:
evt = json.loads(data_str.decode("utf-8", errors="replace"))
if "usage" in evt:
usage_data = evt["usage"]
except Exception:
continue
delta = None
try:
delta = evt["choices"][0]["delta"].get("content")
except Exception:
delta = None
if delta:
delta = fix_mojibake(delta)
reply_parts.append(delta)
sys.stdout.write(delta)
sys.stdout.flush()
reply = "".join(reply_parts).strip()
if usage_data:
prompt_t = int(usage_data.get("prompt_tokens", 0) or 0)
comp_t = int(usage_data.get("completion_tokens", 0) or 0)
total_t = int(usage_data.get("total_tokens", 0) or 0)
usage_obj = {
"prompt_tokens": prompt_t,
"completion_tokens": comp_t,
"total_tokens": total_t
}
reply = fix_mojibake(reply)
if not reply:
sys.stderr.write(get_ui_text(cfg, "o_OpenRouter_Empty_Stream") + "\n")
sys.exit(1)
except requests.exceptions.Timeout:
sys.stderr.write(get_ui_text(cfg, "o_Request_Timeout") + "\n")
sys.exit(1)
except requests.exceptions.ConnectionError:
if is_local_model:
try:
provider, _actual_model = _split_local_model(model_name)
except Exception:
provider = "local"
provider_cfg = local_providers.get(provider, {}) if isinstance(local_providers, dict) else {}
base_url = str(provider_cfg.get("base_url") or "").strip()
custom_error = str(provider_cfg.get("system_error") or "").strip()
# custom error varsa onu kullan
if custom_error:
msg = custom_error.replace(
"{base_url}",
base_url or "http://127.0.0.1:11434"
)
else:
# default local error
msg = get_ui_text(
cfg,
"o_Local_Connection_Error",
provider=provider,
base_url=base_url or "http://127.0.0.1:11434"
)
# chat formatında döndür
result_obj = {
"type": "text",
"content": msg
}
sys.stdout.write(json.dumps(result_obj, ensure_ascii=False))
sys.stdout.flush()
else:
# 🌐 online model
result_obj = {
"type": "text",
"content": get_ui_text(cfg, "o_Internet_Connection_Error")
}
sys.stdout.write(json.dumps(result_obj, ensure_ascii=False))
sys.stdout.flush()
sys.exit(1)
except Exception as e:
sys.stderr.write(get_ui_text(cfg, "o_AI_Error", error=str(e)) + "\n")
sys.exit(1)
# ---------------- SAVE ASSISTANT MEMORY ----------------
if context_mode == "rag":
try:
assistant_text = ""
# image model / local model bazen JSON string döndürüyor
try:
parsed_reply = json.loads(reply)
if isinstance(parsed_reply, dict):
assistant_text = str(parsed_reply.get("content") or "").strip()
except Exception:
assistant_text = str(reply or "").strip()
if assistant_text:
add_memory_chunk(
chat_data,
assistant_text[:memory_chunk_max_chars],
source="chat:assistant"
)
if use_summary and should_update_summary(chat_data, every=summary_update_every):
chat_data["summary"] = make_simple_summary(chat_data)[:summary_max_chars]
save_chat_data(chat_file, chat_data)
except Exception:
pass
sys.exit(0)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink