575 lines
26 KiB
Python
575 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import sqlite3
|
|
import re
|
|
import hashlib
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
from fastapi import UploadFile
|
|
|
|
from backend.app.config import settings
|
|
from backend.app.utils.slug import slugify
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
class DuplicateSourceError(ValueError):
|
|
def __init__(self, existing_source: dict[str, Any]):
|
|
self.existing_source = existing_source
|
|
super().__init__(f"El archivo ya existe en esta semana: {existing_source['original_name']}")
|
|
|
|
|
|
class KnowledgeManager:
|
|
def __init__(self, data_dir: Path | None = None, database_path: Path | None = None):
|
|
self.data_dir = data_dir or settings.app_data_dir
|
|
self.database_path = database_path or settings.database_path
|
|
self.subjects_dir = self.data_dir / "subjects"
|
|
self.tmp_dir = self.data_dir / "tmp"
|
|
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
self.subjects_dir.mkdir(parents=True, exist_ok=True)
|
|
self.tmp_dir.mkdir(parents=True, exist_ok=True)
|
|
self.database_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._init_db()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(self.database_path)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def _init_db(self) -> None:
|
|
with self._connect() as conn:
|
|
conn.executescript(
|
|
"""
|
|
create table if not exists subjects (
|
|
id integer primary key autoincrement,
|
|
name text not null,
|
|
slug text not null unique,
|
|
created_at text not null
|
|
);
|
|
|
|
create table if not exists weeks (
|
|
id integer primary key autoincrement,
|
|
subject_id integer not null,
|
|
number integer not null,
|
|
title text,
|
|
created_at text not null,
|
|
unique(subject_id, number),
|
|
foreign key(subject_id) references subjects(id)
|
|
);
|
|
|
|
create table if not exists sources (
|
|
id integer primary key autoincrement,
|
|
week_id integer not null,
|
|
original_name text not null,
|
|
stored_path text not null,
|
|
source_type text not null,
|
|
status text not null,
|
|
file_size integer,
|
|
sha256 text,
|
|
mime_type text,
|
|
processor text,
|
|
page_ranges text,
|
|
processed_at text,
|
|
created_at text not null,
|
|
foreign key(week_id) references weeks(id)
|
|
);
|
|
|
|
create table if not exists documents (
|
|
id integer primary key autoincrement,
|
|
source_id integer not null,
|
|
title text not null,
|
|
markdown_path text not null,
|
|
created_at text not null,
|
|
foreign key(source_id) references sources(id)
|
|
);
|
|
|
|
create table if not exists jobs (
|
|
id integer primary key autoincrement,
|
|
status text not null,
|
|
message text not null,
|
|
source_id integer,
|
|
document_id integer,
|
|
created_at text not null,
|
|
updated_at text not null,
|
|
foreign key(source_id) references sources(id),
|
|
foreign key(document_id) references documents(id)
|
|
);
|
|
"""
|
|
)
|
|
self._ensure_columns(
|
|
conn,
|
|
"sources",
|
|
{
|
|
"file_size": "integer",
|
|
"sha256": "text",
|
|
"mime_type": "text",
|
|
"processor": "text",
|
|
"page_ranges": "text",
|
|
"processed_at": "text",
|
|
},
|
|
)
|
|
self._backfill_source_file_metadata(conn)
|
|
|
|
def _ensure_columns(self, conn: sqlite3.Connection, table: str, columns: dict[str, str]) -> None:
|
|
existing = {row["name"] for row in conn.execute(f"pragma table_info({table})").fetchall()}
|
|
for name, definition in columns.items():
|
|
if name not in existing:
|
|
conn.execute(f"alter table {table} add column {name} {definition}")
|
|
|
|
def _backfill_source_file_metadata(self, conn: sqlite3.Connection) -> None:
|
|
rows = conn.execute("select id, stored_path from sources where sha256 is null or file_size is null").fetchall()
|
|
for row in rows:
|
|
path = Path(row["stored_path"])
|
|
if not path.exists():
|
|
continue
|
|
sha256, file_size = self._file_hash_and_size(path)
|
|
conn.execute("update sources set sha256 = ?, file_size = ? where id = ?", (sha256, file_size, row["id"]))
|
|
|
|
def create_subject(self, name: str) -> dict[str, Any]:
|
|
base_slug = slugify(name)
|
|
slug = base_slug
|
|
suffix = 2
|
|
with self._connect() as conn:
|
|
while conn.execute("select 1 from subjects where slug = ?", (slug,)).fetchone():
|
|
slug = f"{base_slug}-{suffix}"
|
|
suffix += 1
|
|
now = utc_now()
|
|
cursor = conn.execute(
|
|
"insert into subjects (name, slug, created_at) values (?, ?, ?)",
|
|
(name, slug, now),
|
|
)
|
|
(self.subjects_dir / slug).mkdir(parents=True, exist_ok=True)
|
|
return self.get_subject(cursor.lastrowid)
|
|
|
|
def get_subject(self, subject_id: int) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
row = conn.execute("select * from subjects where id = ?", (subject_id,)).fetchone()
|
|
if not row:
|
|
raise ValueError("Materia no encontrada")
|
|
return dict(row)
|
|
|
|
def update_subject(self, subject_id: int, name: str) -> dict[str, Any]:
|
|
if not name.strip():
|
|
raise ValueError("El nombre de la materia no puede estar vacio")
|
|
self.get_subject(subject_id)
|
|
with self._connect() as conn:
|
|
conn.execute("update subjects set name = ? where id = ?", (name.strip(), subject_id))
|
|
return self.get_subject(subject_id)
|
|
|
|
def delete_subject(self, subject_id: int) -> None:
|
|
subject = self.get_subject(subject_id)
|
|
with self._connect() as conn:
|
|
week_rows = conn.execute("select id from weeks where subject_id = ?", (subject_id,)).fetchall()
|
|
week_ids = [row["id"] for row in week_rows]
|
|
for week_id in week_ids:
|
|
source_rows = conn.execute("select id from sources where week_id = ?", (week_id,)).fetchall()
|
|
source_ids = [row["id"] for row in source_rows]
|
|
for source_id in source_ids:
|
|
conn.execute("delete from documents where source_id = ?", (source_id,))
|
|
conn.execute("delete from jobs where source_id = ?", (source_id,))
|
|
conn.execute("delete from sources where week_id = ?", (week_id,))
|
|
conn.execute("delete from weeks where subject_id = ?", (subject_id,))
|
|
conn.execute("delete from subjects where id = ?", (subject_id,))
|
|
shutil.rmtree(self.subjects_dir / subject["slug"], ignore_errors=True)
|
|
|
|
def list_subjects(self) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute("select * from subjects order by created_at desc").fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def create_week(self, subject_id: int, number: int, title: str | None = None) -> dict[str, Any]:
|
|
subject = self.get_subject(subject_id)
|
|
now = utc_now()
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"insert or ignore into weeks (subject_id, number, title, created_at) values (?, ?, ?, ?)",
|
|
(subject_id, number, title, now),
|
|
)
|
|
if title:
|
|
conn.execute("update weeks set title = ? where subject_id = ? and number = ?", (title, subject_id, number))
|
|
row = conn.execute(
|
|
"select * from weeks where subject_id = ? and number = ?",
|
|
(subject_id, number),
|
|
).fetchone()
|
|
self.week_dir(subject["slug"], number).mkdir(parents=True, exist_ok=True)
|
|
for child in ("raw", "processed", "exports"):
|
|
(self.week_dir(subject["slug"], number) / child).mkdir(parents=True, exist_ok=True)
|
|
return dict(row)
|
|
|
|
def list_weeks(self, subject_id: int) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute("select * from weeks where subject_id = ? order by number", (subject_id,)).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def delete_week(self, subject_id: int, number: int) -> None:
|
|
subject = self.get_subject(subject_id)
|
|
week = self.get_week(subject_id, number)
|
|
with self._connect() as conn:
|
|
source_rows = conn.execute("select id from sources where week_id = ?", (week["id"],)).fetchall()
|
|
for row in source_rows:
|
|
conn.execute("delete from documents where source_id = ?", (row["id"],))
|
|
conn.execute("delete from jobs where source_id = ?", (row["id"],))
|
|
conn.execute("delete from sources where week_id = ?", (week["id"],))
|
|
conn.execute("delete from weeks where id = ?", (week["id"],))
|
|
shutil.rmtree(self.week_dir(subject["slug"], number), ignore_errors=True)
|
|
|
|
def get_week(self, subject_id: int, number: int) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"select * from weeks where subject_id = ? and number = ?",
|
|
(subject_id, number),
|
|
).fetchone()
|
|
if not row:
|
|
raise ValueError("Semana no encontrada")
|
|
return dict(row)
|
|
|
|
def update_week(self, subject_id: int, current_number: int, new_number: int, title: str | None = None) -> dict[str, Any]:
|
|
subject = self.get_subject(subject_id)
|
|
week = self.get_week(subject_id, current_number)
|
|
old_dir: Path | None = None
|
|
new_dir: Path | None = None
|
|
if new_number != current_number:
|
|
with self._connect() as conn:
|
|
existing = conn.execute(
|
|
"select 1 from weeks where subject_id = ? and number = ?",
|
|
(subject_id, new_number),
|
|
).fetchone()
|
|
if existing:
|
|
raise ValueError("Ya existe una semana con ese numero")
|
|
|
|
old_dir = self.week_dir(subject["slug"], current_number)
|
|
new_dir = self.week_dir(subject["slug"], new_number)
|
|
if old_dir.exists() and new_dir.exists():
|
|
raise ValueError("Ya existe una carpeta para ese numero de semana")
|
|
if old_dir.exists():
|
|
old_dir.rename(new_dir)
|
|
else:
|
|
for child in ("raw", "processed", "exports"):
|
|
(new_dir / child).mkdir(parents=True, exist_ok=True)
|
|
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"update weeks set number = ?, title = ? where id = ?",
|
|
(new_number, title, week["id"]),
|
|
)
|
|
if old_dir and new_dir:
|
|
old_prefix = str(old_dir)
|
|
new_prefix = str(new_dir)
|
|
conn.execute(
|
|
"update sources set stored_path = replace(stored_path, ?, ?) where week_id = ? and stored_path like ?",
|
|
(old_prefix, new_prefix, week["id"], f"{old_prefix}%"),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
update documents
|
|
set markdown_path = replace(markdown_path, ?, ?)
|
|
where source_id in (select id from sources where week_id = ?)
|
|
and markdown_path like ?
|
|
""",
|
|
(old_prefix, new_prefix, week["id"], f"{old_prefix}%"),
|
|
)
|
|
return self.get_week(subject_id, new_number)
|
|
|
|
def week_dir(self, subject_slug: str, week_number: int) -> Path:
|
|
return self.subjects_dir / subject_slug / f"week-{week_number:02d}"
|
|
|
|
def save_upload(self, subject_id: int, week_number: int, upload: UploadFile) -> dict[str, Any]:
|
|
subject = self.get_subject(subject_id)
|
|
week = self.get_week(subject_id, week_number)
|
|
filename = Path(upload.filename or "archivo").name
|
|
raw_dir = self.week_dir(subject["slug"], week_number) / "raw"
|
|
raw_dir.mkdir(parents=True, exist_ok=True)
|
|
stored_path = self._unique_path(raw_dir / filename)
|
|
digest = hashlib.sha256()
|
|
file_size = 0
|
|
with stored_path.open("wb") as out_file:
|
|
while chunk := upload.file.read(1024 * 1024):
|
|
file_size += len(chunk)
|
|
digest.update(chunk)
|
|
out_file.write(chunk)
|
|
sha256 = digest.hexdigest()
|
|
source_type = self.detect_source_type(stored_path)
|
|
now = utc_now()
|
|
with self._connect() as conn:
|
|
duplicate = conn.execute(
|
|
"select * from sources where week_id = ? and sha256 = ? limit 1",
|
|
(week["id"], sha256),
|
|
).fetchone()
|
|
if duplicate:
|
|
stored_path.unlink(missing_ok=True)
|
|
raise DuplicateSourceError(dict(duplicate))
|
|
cursor = conn.execute(
|
|
"""
|
|
insert into sources (week_id, original_name, stored_path, source_type, status, file_size, sha256, mime_type, created_at)
|
|
values (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(week["id"], filename, str(stored_path), source_type, "stored", file_size, sha256, upload.content_type, now),
|
|
)
|
|
return self.get_source(cursor.lastrowid)
|
|
|
|
def _file_hash_and_size(self, path: Path) -> tuple[str, int]:
|
|
digest = hashlib.sha256()
|
|
file_size = 0
|
|
with path.open("rb") as file:
|
|
while chunk := file.read(1024 * 1024):
|
|
file_size += len(chunk)
|
|
digest.update(chunk)
|
|
return digest.hexdigest(), file_size
|
|
|
|
def create_job(self, source_id: int, message: str = "En cola") -> dict[str, Any]:
|
|
now = utc_now()
|
|
with self._connect() as conn:
|
|
cursor = conn.execute(
|
|
"insert into jobs (status, message, source_id, created_at, updated_at) values (?, ?, ?, ?, ?)",
|
|
("queued", message, source_id, now, now),
|
|
)
|
|
return self.get_job(cursor.lastrowid)
|
|
|
|
def update_job(self, job_id: int, status: str, message: str, document_id: int | None = None) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"update jobs set status = ?, message = ?, document_id = coalesce(?, document_id), updated_at = ? where id = ?",
|
|
(status, message, document_id, utc_now(), job_id),
|
|
)
|
|
|
|
def update_source_status(self, source_id: int, status: str) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute("update sources set status = ? where id = ?", (status, source_id))
|
|
|
|
def source_has_active_job(self, source_id: int) -> bool:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"select 1 from jobs where source_id = ? and status in ('queued', 'processing') limit 1",
|
|
(source_id,),
|
|
).fetchone()
|
|
return bool(row)
|
|
|
|
def get_job(self, job_id: int) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
row = conn.execute("select * from jobs where id = ?", (job_id,)).fetchone()
|
|
if not row:
|
|
raise ValueError("Trabajo no encontrado")
|
|
return dict(row)
|
|
|
|
def list_jobs_for_week(self, subject_id: int, week_number: int) -> list[dict[str, Any]]:
|
|
week = self.get_week(subject_id, week_number)
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
select jobs.* from jobs
|
|
join sources on sources.id = jobs.source_id
|
|
where sources.week_id = ?
|
|
order by jobs.created_at desc
|
|
""",
|
|
(week["id"],),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def get_source(self, source_id: int) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
row = conn.execute("select * from sources where id = ?", (source_id,)).fetchone()
|
|
if not row:
|
|
raise ValueError("Fuente no encontrada")
|
|
return dict(row)
|
|
|
|
def list_sources(self, subject_id: int, week_number: int) -> list[dict[str, Any]]:
|
|
week = self.get_week(subject_id, week_number)
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
select sources.*, documents.id as document_id, documents.title as document_title,
|
|
documents.markdown_path as document_markdown_path, documents.created_at as document_created_at
|
|
from sources
|
|
left join documents on documents.source_id = sources.id
|
|
where sources.week_id = ?
|
|
order by sources.created_at desc
|
|
""",
|
|
(week["id"],),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def delete_source(self, source_id: int) -> None:
|
|
source = self.get_source(source_id)
|
|
with self._connect() as conn:
|
|
doc_rows = conn.execute("select markdown_path from documents where source_id = ?", (source_id,)).fetchall()
|
|
conn.execute("delete from documents where source_id = ?", (source_id,))
|
|
conn.execute("delete from jobs where source_id = ?", (source_id,))
|
|
conn.execute("delete from sources where id = ?", (source_id,))
|
|
Path(source["stored_path"]).unlink(missing_ok=True)
|
|
for row in doc_rows:
|
|
Path(row["markdown_path"]).unlink(missing_ok=True)
|
|
|
|
def create_document(self, source_id: int, title: str, markdown: str, processor: str | None = None, page_ranges: str | None = None) -> dict[str, Any]:
|
|
source = self.get_source(source_id)
|
|
week_context = self.get_week_context(source["week_id"])
|
|
processed_dir = self.week_dir(week_context["subject_slug"], week_context["week_number"]) / "processed"
|
|
processed_dir.mkdir(parents=True, exist_ok=True)
|
|
markdown_path = self._unique_path(processed_dir / f"{slugify(title)}.md")
|
|
markdown_path.write_text(self._sanitize_markdown(markdown), encoding="utf-8")
|
|
now = utc_now()
|
|
with self._connect() as conn:
|
|
old_docs = conn.execute("select markdown_path from documents where source_id = ?", (source_id,)).fetchall()
|
|
conn.execute("delete from jobs where document_id in (select id from documents where source_id = ?)", (source_id,))
|
|
conn.execute("delete from documents where source_id = ?", (source_id,))
|
|
cursor = conn.execute(
|
|
"insert into documents (source_id, title, markdown_path, created_at) values (?, ?, ?, ?)",
|
|
(source_id, title, str(markdown_path), now),
|
|
)
|
|
conn.execute(
|
|
"update sources set status = ?, processor = ?, page_ranges = ?, processed_at = ? where id = ?",
|
|
("processed", processor, page_ranges, now, source_id),
|
|
)
|
|
for row in old_docs:
|
|
Path(row["markdown_path"]).unlink(missing_ok=True)
|
|
return self.get_document(cursor.lastrowid)
|
|
|
|
def get_document(self, document_id: int) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
row = conn.execute("select * from documents where id = ?", (document_id,)).fetchone()
|
|
if not row:
|
|
raise ValueError("Documento no encontrado")
|
|
return dict(row)
|
|
|
|
def update_document_markdown(self, document_id: int, content: str) -> dict[str, Any]:
|
|
document = self.get_document(document_id)
|
|
Path(document["markdown_path"]).write_text(self._sanitize_markdown(content), encoding="utf-8")
|
|
return self.get_document(document_id)
|
|
|
|
def delete_document(self, document_id: int) -> None:
|
|
document = self.get_document(document_id)
|
|
Path(document["markdown_path"]).unlink(missing_ok=True)
|
|
with self._connect() as conn:
|
|
conn.execute("delete from jobs where document_id = ?", (document_id,))
|
|
conn.execute("delete from documents where id = ?", (document_id,))
|
|
|
|
def list_documents(self, subject_id: int, week_number: int) -> list[dict[str, Any]]:
|
|
week = self.get_week(subject_id, week_number)
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
select documents.*, sources.original_name as source_original_name,
|
|
sources.source_type as source_type, sources.status as source_status,
|
|
sources.processor as source_processor, sources.page_ranges as source_page_ranges
|
|
from documents
|
|
join sources on sources.id = documents.source_id
|
|
where sources.week_id = ?
|
|
order by documents.created_at desc
|
|
""",
|
|
(week["id"],),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def get_week_context(self, week_id: int) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
select weeks.id as week_id, weeks.number as week_number, weeks.title as week_title,
|
|
subjects.id as subject_id, subjects.name as subject_name, subjects.slug as subject_slug
|
|
from weeks join subjects on subjects.id = weeks.subject_id
|
|
where weeks.id = ?
|
|
""",
|
|
(week_id,),
|
|
).fetchone()
|
|
if not row:
|
|
raise ValueError("Contexto de semana no encontrado")
|
|
return dict(row)
|
|
|
|
def export_week_markdown(self, subject_id: int, week_number: int) -> Path:
|
|
subject = self.get_subject(subject_id)
|
|
docs = self.list_documents(subject_id, week_number)
|
|
export_dir = self.week_dir(subject["slug"], week_number) / "exports"
|
|
export_dir.mkdir(parents=True, exist_ok=True)
|
|
export_path = export_dir / f"{subject['slug']}-week-{week_number:02d}.md"
|
|
parts = [
|
|
"---\n"
|
|
f"subject: {subject['name']}\n"
|
|
f"subject_slug: {subject['slug']}\n"
|
|
f"week: {week_number}\n"
|
|
f"documents: {len(docs)}\n"
|
|
f"exported_at: {utc_now()}\n"
|
|
"---\n\n"
|
|
f"# {subject['name']} - Semana {week_number}\n"
|
|
]
|
|
for doc in docs:
|
|
path = Path(doc["markdown_path"])
|
|
content = self._strip_frontmatter(self._read_markdown(path)).strip()
|
|
parts.append(f"<!-- document_id: {doc['id']} -->\n\n{content}\n")
|
|
export_path.write_text(self._sanitize_markdown("\n\n---\n\n".join(parts)), encoding="utf-8")
|
|
return export_path
|
|
|
|
def export_week_markdown_zip(self, subject_id: int, week_number: int) -> Path:
|
|
subject = self.get_subject(subject_id)
|
|
docs = self.list_documents(subject_id, week_number)
|
|
export_dir = self.week_dir(subject["slug"], week_number) / "exports"
|
|
export_dir.mkdir(parents=True, exist_ok=True)
|
|
export_path = export_dir / f"{subject['slug']}-week-{week_number:02d}-documentos.zip"
|
|
used_names: set[str] = set()
|
|
with zipfile.ZipFile(export_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
|
|
for index, doc in enumerate(docs, start=1):
|
|
path = Path(doc["markdown_path"])
|
|
name = self._unique_archive_name(f"{index:02d}-{slugify(doc['title'])}.md", used_names)
|
|
archive.writestr(name, self._read_markdown(path))
|
|
return export_path
|
|
|
|
def _unique_archive_name(self, filename: str, used_names: set[str]) -> str:
|
|
path = Path(filename)
|
|
candidate = path.name
|
|
suffix = path.suffix
|
|
stem = path.stem
|
|
counter = 2
|
|
while candidate in used_names:
|
|
candidate = f"{stem}-{counter}{suffix}"
|
|
counter += 1
|
|
used_names.add(candidate)
|
|
return candidate
|
|
|
|
def _read_markdown(self, path: Path) -> str:
|
|
return self._sanitize_markdown(path.read_text(encoding="utf-8", errors="replace"))
|
|
|
|
def _sanitize_markdown(self, content: str) -> str:
|
|
return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", content)
|
|
|
|
def _strip_frontmatter(self, content: str) -> str:
|
|
lines = content.splitlines()
|
|
if not lines or lines[0].strip() != "---":
|
|
return content
|
|
for index, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
return "\n".join(lines[index + 1 :]).lstrip()
|
|
return content
|
|
|
|
def detect_source_type(self, path: Path) -> str:
|
|
ext = path.suffix.lower().lstrip(".")
|
|
if ext in {"txt", "md"}:
|
|
return "text"
|
|
if ext == "docx":
|
|
return "docx"
|
|
if ext == "pdf":
|
|
return "pdf"
|
|
if ext in {"mp3", "wav", "m4a", "ogg", "flac", "webm"}:
|
|
return "audio"
|
|
if ext in {"mp4", "mov", "mkv", "avi"}:
|
|
return "video"
|
|
return "unknown"
|
|
|
|
def _unique_path(self, path: Path) -> Path:
|
|
if not path.exists():
|
|
return path
|
|
stem = path.stem
|
|
suffix = path.suffix
|
|
for index in range(2, 10_000):
|
|
candidate = path.with_name(f"{stem}-{index}{suffix}")
|
|
if not candidate.exists():
|
|
return candidate
|
|
raise RuntimeError("No se pudo generar un nombre de archivo unico")
|
|
|
|
|
|
def rows_to_models(rows: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return [dict(row) for row in rows]
|