Merge pull request #57

v3.3.0

acacore/__version__.py

@@ -1 +1 @@
-__version__ = "3.2.0"
+__version__ = "3.3.0"

acacore/database/upgrade.py

@@ -1,6 +1,8 @@
 from functools import reduce
 from json import dumps
+from json import JSONDecodeError
 from json import loads
+from pathlib import Path
 from sqlite3 import Connection
 from sqlite3 import DatabaseError
 from typing import Any
@@ -15,7 +17,6 @@
     "is_latest",
 ]
 
-
 from .files_db import FileDB
 
 
@@ -214,6 +215,75 @@ def convert_action_data(data: dict) -> dict:
     return set_db_version(conn, Version("3.2.0"))
 
 
+# noinspection SqlResolve
+def upgrade_3_2to3_3(conn: Connection) -> Version:
+    if not conn.execute("select 1 from pragma_table_info('Files') where name = 'original_name'").fetchone():
+        conn.execute("alter table Files add column original_name text not null default ''")
+    if not conn.execute("select 1 from pragma_table_info('Files') where name = 'processed_name'").fetchone():
+        conn.execute("alter table Files add column processed_name text default '[]'")
+
+    def _find_original_name(uuid: str, relative_path: str) -> str:
+        original_path: Path = Path(relative_path)
+        original_name: str = original_path.name
+
+        for [data_raw] in conn.execute(
+            "select data from History"
+            " where uuid = ? and data is not null and data not in ('', '\"\"', 'null', '[]', '{}')"
+            " order by time desc",
+            (uuid,),
+        ):
+            try:
+                data = loads(data_raw)
+            except JSONDecodeError:
+                continue
+
+            if (
+                not isinstance(data, list)
+                or len(data) != 2
+                or not isinstance(data[0], str)
+                or not isinstance(data[1], str)
+            ):
+                continue
+
+            a: str = data[0]
+            b: str = data[1]
+
+            if a == str(original_path):
+                original_path = Path(b)
+                original_name = original_path.name
+            elif b == str(original_path):
+                original_path = Path(a)
+                original_name = original_path.name
+            elif a == original_name:
+                original_name = b
+                original_path = original_path.with_name(original_name)
+            elif b == original_name:
+                original_name = a
+                original_path = original_path.with_name(original_name)
+
+        return original_name
+
+    conn.executemany(
+        "update Files set original_name = ? where uuid = ?",
+        (
+            (_find_original_name(uuid, relative_path), uuid)
+            for uuid, relative_path in conn.execute(
+                "select distinct f.uuid, f.relative_path from Files f join History h where h.uuid = f.uuid"
+            )
+        ),
+    )
+
+    conn.executemany(
+        "update Files set original_name = ? where uuid = ?",
+        (
+            (Path(relative_path).name, uuid)
+            for uuid, relative_path in conn.execute("select uuid, relative_path from Files where original_name = ''")
+        ),
+    )
+
+    return set_db_version(conn, Version("3.3.0"))
+
+
 def get_upgrade_function(current_version: Version, latest_version: Version) -> Callable[[Connection], Version]:
     if current_version < Version("2.0.0"):
         return upgrade_1to2
@@ -229,6 +299,8 @@ def get_upgrade_function(current_version: Version, latest_version: Version) -> C
         return upgrade_3_0_6to3_0_7
     elif current_version < Version("3.2.0"):
         return upgrade_3_1to3_2
+    elif current_version < Version("3.3.0"):
+        return upgrade_3_2to3_3
     elif current_version < latest_version:
         return lambda c: set_db_version(c, Version(__version__))
     else:

acacore/models/file.py

@@ -6,6 +6,7 @@
 from pydantic import BaseModel
 from pydantic import Field
 from pydantic import field_validator
+from pydantic import model_validator
 from pydantic import UUID4
 
 from acacore.database.column import DBField
@@ -89,8 +90,18 @@ class File(BaseModel):
     parent: UUID4 | None = None
     processed: bool = False
     lock: bool = False
+    original_name: str
+    processed_names: list[str] = Field(default_factory=list)
     root: Path | None = DBField(None, ignore=True)
 
+    # noinspection PyNestedDecorators
+    @model_validator(mode="before")
+    @classmethod
+    def _model_validator(cls, data: dict):
+        if isinstance(data, dict):
+            data["original_name"] = data.get("original_name", "").strip() or Path(data["relative_path"]).name
+        return data
+
     # noinspection PyNestedDecorators
     @field_validator("action_data", mode="before")
     @classmethod

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "acacore"
-version = "3.2.0"
+version = "3.3.0"
 description = ""
 authors = ["Matteo Campinoti <[email protected]>"]
 license = "GPL-3.0"

tests/test_database.py

@@ -26,7 +26,7 @@
 from acacore.models.reference_files import ConvertAction
 
 
-@pytest.fixture()
+@pytest.fixture
 def database_path(temp_folder: Path) -> Path:
     path: Path = temp_folder / "files.db"
     path.unlink(missing_ok=True)

tests/test_siegfried.py

@@ -7,12 +7,12 @@
 from acacore.siegfried import Siegfried
 
 
-@pytest.fixture()
+@pytest.fixture
 def siegfried_folder(test_folder: Path) -> Path:
     return test_folder / "siegfried"
 
 
-@pytest.fixture()
+@pytest.fixture
 def siegfried(siegfried_folder: Path) -> Siegfried:
     return Siegfried(Path(environ["GOPATH"], "bin", "sf"), "pronom.sig", siegfried_folder)