fuck orm

db/__init__.py

@@ -1,4 +1 @@
-from .base import init
+from .quest import QuestDB
-from .quest import Quest, QuestScene, QuestPart
-from .resource import ResourceEntry
-from .base import session

db/base.py

@@ -1,15 +1,5 @@
 import os
-
+import sqlite3
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy import text
-from sqlalchemy.ext.declarative import declarative_base
-
-Base = declarative_base()
-
-engine = create_engine("sqlite:///:memory:", echo=False)
-session = sessionmaker(bind=engine, autoflush=False, autocommit=False)()
-session.flush = lambda: None
 
 DATABASES = {
     "QuestMst.db": "Quest",
@@ -19,11 +9,28 @@ DATABASES = {
 }
 
 
-def init(path):
+class BaseDB(object):
-    for d in DATABASES:
+    conn = None
-        if not os.path.isfile(os.path.join(path, d)):
+    cursor = None
-            raise FileNotFoundError("Database {} not found".format(d))
 
-    for d in DATABASES:
+    def __init__(self, path, databases):
-        t = text("attach database :path as :schema")
+        self.conn = sqlite3.connect(":memory:")
-        engine.execute(t, path=os.path.join(path, d), schema=DATABASES[d])
+        self.conn.row_factory = sqlite3.Row
+        self.cursor = self.conn.cursor()
+        self.attach_dbs(databases, path)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.quest_conn.close()
+
+    def attach_dbs(self, databases, path):
+        for d in databases:
+            if not os.path.isfile(os.path.join(path, d)):
+                raise FileNotFoundError(f"Database {d} not found")
+
+        for d in databases.items():
+            self.cursor.execute(
+                "ATTACH DATABASE ? AS ?", (os.path.join(path, d[0]), d[1])
+            )

db/quest.py

@@ -1,76 +1,74 @@
-from sqlalchemy import Column, Integer, BigInteger, String, ForeignKey
+from .base import BaseDB
-from sqlalchemy.orm import relationship
 
-from .base import Base
+DATABASES = {
+    "QuestMst.db": "Quest",
+    "QuestSceneMst.db": "QuestScene",
+    "QuestPartMst.db": "QuestPart",
+    "ResourceEntry.db": "ResourceEntry",
+}
 
 
-class Quest(Base):
+class QuestDB(BaseDB):
-    __tablename__ = "QuestMstRecord"
+    def __init__(self, path):
-    __table_args__ = {"schema": "Quest"}
+        super().__init__(path, DATABASES)
 
-    questMstId = Column(Integer, primary_key=True)
+    def get_quests(self):
-    questType = Column(Integer)
+        self.cursor.execute(
-    sortNum = Column(Integer)
+            """
-    name = Column(String)
+            SELECT
-    prevQuestMstId = Column(Integer)
+                q.questMstId,
-    eventItemMstIds = Column(Integer)
+                q.name,
-    valid = Column(Integer)
+                group_concat(DISTINCT qp.questSceneMstId) AS scenes
-    baseQuestMstId = Column(Integer)
+            FROM Quest.QuestMstRecord q
-    updatedTime = Column(BigInteger)
+            INNER JOIN QuestScene.QuestSceneMstRecord qs
-
+            ON q.questMstId = qs.questMstId
-    scenes = relationship("QuestScene")
+            INNER JOIN QuestPart.QuestPartMstRecord qp
-
+            ON qs.questSceneMstId = qp.questSceneMstId
-    def __repr__(self):
+            WHERE q.baseQuestMstId = 0
-        return "<Quest(questMstId={}, name={})>".format(self.questMstId, self.name)
+            AND (qp.beforeTalkName != "" OR qp.afterTalkName != "")
-
+            GROUP BY q.questMstId
-
+            """
-class QuestScene(Base):
-    __tablename__ = "QuestSceneMstRecord"
-    __table_args__ = {"schema": "QuestScene"}
-
-    questSceneMstId = Column(Integer, primary_key=True)
-    questMstId = Column(Integer, ForeignKey("Quest.QuestMstRecord.questMstId"))
-    _type = Column("type", Integer)
-    sortNum = Column(Integer)
-    name = Column(String)
-    _filter = Column("filter", Integer)
-    prevQuestSceneMstId = Column(
-        Integer, ForeignKey("QuestScene.QuestSceneMstRecord.questSceneMstId")
         )
-    releaseSerial = Column(Integer)
+        res = self.cursor.fetchall()
-    releaseEvolutionLevel = Column(Integer)
+        quests = {}
-    summaryText = Column(String)
+        for quest_id, name, scenes in res:
-    presentType = Column(String)
+            quests[quest_id] = {}
-    objectId = Column(Integer)
+            quests[quest_id]["Name"] = name
-    num = Column(Integer)
+            quests[quest_id]["Scenes"] = scenes.split(",")
-    appearanceType = Column(Integer)
+        return quests
-    isPrologue = Column(Integer)
-    isEpilogue = Column(Integer)
-    viewType = Column(Integer)
-    updatedTime = Column(BigInteger)
 
-    previous = relationship("QuestScene", remote_side=[questSceneMstId])
+    def get_scenes(self):
-    parts = relationship("QuestPart")
+        self.cursor.execute(
-
+            """
-
+            SELECT
-class QuestPart(Base):
+                questSceneMstId,
-    __tablename__ = "QuestPartMstRecord"
+                name,
-    __table_args__ = {"schema": "QuestPart"}
+                summaryText,
-
+                group_concat(partIds) as parts
-    questSceneMstId = Column(
+            FROM (
-        Integer,
+                    SELECT
-        ForeignKey("QuestScene.QuestSceneMstRecord.questSceneMstId"),
+                        qs.questSceneMstId,
-        primary_key=True,
+                        qs.name,
-    )  # now this is pod racing
+                        qs.summaryText,
-    partNum = Column(Integer, primary_key=True)
+                        CASE
-    waveNum = Column(Integer)
+                            WHEN qp.afterTalkName == '' AND qp.beforeTalkName == '' THEN NULL
-    stamina = Column(Integer)
+                            WHEN qp.afterTalkName == '' AND qp.beforeTalkName != '' THEN qp.beforeTalkName
-    exp = Column(Integer)
+                            WHEN qp.afterTalkName != '' AND qp.beforeTalkName == '' THEN qp.afterTalkName
-    expertPoint = Column(Integer)
+                            ELSE qp.beforeTalkName || ',' || qp.afterTalkName
-    recommendLevel = Column(Integer)
+                        END AS partIds
-    beforeTalkName = Column(String)
+                    FROM QuestScene.QuestSceneMstRecord qs
-    afterTalkName = Column(String)
+                    INNER JOIN QuestPart.QuestPartMstRecord qp
-    battleBackgroundImg = Column(String)
+                    ON qs.questSceneMstId = qp.questSceneMstId
-    musicMstId = Column(Integer)
+                    WHERE (qp.beforeTalkName != "" OR qp.afterTalkName != "")
-    isFixedDeck = Column(Integer)
+            )
-    updatedTime = Column(BigInteger)
+            GROUP BY questSceneMstId
+            """
+        )
+        res = self.cursor.fetchall()
+        scenes = {}
+        for scene_id, name, summary, parts in res:
+            scenes[scene_id] = {}
+            scenes[scene_id]["Name"] = name
+            scenes[scene_id]["SummaryText"] = summary
+            scenes[scene_id]["Parts"] = parts.split(",")
+        return scenes

db/resource.py

@@ -1,14 +0,0 @@
-from sqlalchemy import Column, BigInteger, String
-
-from .base import Base
-
-
-class ResourceEntry(Base):
-    __tablename__ = "ResourceEntryRecord"
-    __table_args__ = {"schema": "ResourceEntry"}
-
-    path = Column(String, primary_key=True)
-    serverPath = Column(String)
-    localPath = Column(String)
-    digest = Column(String)
-    fileSize = Column(BigInteger)

diva/quest.py

@@ -3,153 +3,92 @@ import io
 import json
 import os
 
-from sqlalchemy.orm import contains_eager
-
 
 def update_missions(dir_in, old_path, languages):
-    db.init(dir_in)
+    qdb = db.QuestDB(dir_in)
-    quest_mst(old_path, languages)
+    quest_mst(qdb, old_path, languages)
-    scene_mst(old_path, languages)
+    scene_mst(qdb, old_path, languages)
 
 
-def scene_mst(old_path, languages):
+def scene_mst(qdb, old_path, languages):
-    result = (
+    scenes = qdb.get_scenes()
-        db.session.query(db.QuestScene)
-        .join(db.QuestScene.parts)
-        .options(contains_eager(db.QuestScene.parts))
-        .filter(
-            (db.QuestPart.beforeTalkName != "") | (db.QuestPart.afterTalkName != "")
-        )
-        .filter(db.QuestScene.parts.any())
-        .all()
-    )
-
-    scenes = {}
-    for x in result:
-        scenes[str(x.questSceneMstId)] = {
-            "Name": x.name,
-            "SummaryText": x.summaryText,
-            "Parts": [],
-            "Folder": "",
-        }
-
-        id_num = ""
-        # this is slow as balls but i'm way too tired to figure out how to optimize it
-        for y in x.parts:
-            if y.beforeTalkName:
-                id_num = y.beforeTalkName
-                scenes[str(x.questSceneMstId)]["Parts"].append(y.beforeTalkName)
-            if y.afterTalkName:
-                id_num = y.afterTalkName
-                scenes[str(x.questSceneMstId)]["Parts"].append(y.afterTalkName)
-
-            # i checked every entry, moving it to do once for speed
-            if not scenes[str(x.questSceneMstId)]["Folder"] and id_num:
-                path = (
-                    db.session.query(db.ResourceEntry)
-                    .filter(
-                        db.ResourceEntry.path.like(
-                            "%/Scenario/{}.tsv.utage".format(id_num)
-                        )
-                    )
-                    .filter(
-                        db.ResourceEntry.path.notlike("%Utage/side03/%.tsv.utage")
-                    )  # dead folder with dupes too lazy to parse settings tsv
-                    .one()
-                )
-                folder = path.path.split("/")[2]
-                if (
-                    scenes[str(x.questSceneMstId)]["Folder"]
-                    and scenes[str(x.questSceneMstId)]["Folder"] != folder
-                ):
-                    raise KeyError(
-                        "Internal path inconsitency, bailing out {}".format(id_num)
-                    )
-                scenes[str(x.questSceneMstId)]["Folder"] = folder
 
     with io.open(
         os.path.join(old_path, "XduScene.json"), "w", newline="\n"
     ) as json_file:
-        json.dump(scenes, json_file, ensure_ascii=False, indent="\t", sort_keys=True)
+        json.dump(
+            scenes, json_file, ensure_ascii=False, indent="\t", sort_keys=True
+        )
 
-    for l in languages:
+    # TODO: reimplement folder finder
+
+    for lang in languages:
         out_dict = {}
-        if l == "jpn":
+        if lang == "jpn":
-            for x in result:
+            for key, value in scenes.items():
-                out_dict[str(x.questSceneMstId)] = {
+                out_dict[key] = {
-                    "Name": x.name,
+                    "Name": value.name,
-                    "SummaryText": x.summaryText,
+                    "SummaryText": value.summaryText,
                     "Credits": "POKELABO",
                     "Enabled": False,
                 }
         else:
-            for x in result:
+            for key in scenes.keys():
-                out_dict[str(x.questSceneMstId)] = {
+                out_dict[key] = {
                     "Name": "",
                     "SummaryText": "",
                     "Credits": "",
                     "Enabled": False,
                 }
 
-        langfile = os.path.join(old_path, "XduSceneNames_{}.json".format(l))
+        langfile = os.path.join(old_path, f"XduSceneNames_{lang}.json")
         if os.path.isfile(langfile):
             with open(langfile, "r") as lang_file:
                 lang_dict = json.load(lang_file)
                 out_dict.update(lang_dict)
 
-        with io.open(
+        with io.open(langfile, "w", newline="\n") as lang_file:
-            langfile, "w", newline="\n"
-        ) as lang_file:  # you're using git right
             json.dump(
-                out_dict, lang_file, ensure_ascii=False, indent="\t", sort_keys=True
+                out_dict,
+                lang_file,
+                ensure_ascii=False,
+                indent="\t",
+                sort_keys=True,
             )
 
 
-def quest_mst(old_path, languages):
+def quest_mst(qdb, old_path, languages):
-    # i regret my life choices
+    quests = qdb.get_quests()
-    result = (
-        db.session.query(db.Quest)
-        .filter(db.Quest.baseQuestMstId == 0)
-        .join(db.Quest.scenes)
-        .join(db.QuestScene.parts)
-        .options(contains_eager(db.Quest.scenes).contains_eager(db.QuestScene.parts))
-        .filter(db.QuestPart.beforeTalkName != "" or db.QuestPart.afterTalkName != "")
-        .filter(db.QuestScene.parts.any())
-        .filter(db.Quest.scenes.any())
-        .all()
-    )
-    quests = {}
-    for x in result:
-        quests[str(x.questMstId)] = {
-            "Name": x.name,
-            "Scenes": [d.questSceneMstId for d in x.scenes],
-        }
 
     with io.open(
         os.path.join(old_path, "XduQuest.json"), "w", newline="\n"
     ) as json_file:
-        json.dump(quests, json_file, ensure_ascii=False, indent="\t", sort_keys=True)
+        json.dump(
+            quests, json_file, ensure_ascii=False, indent="\t", sort_keys=True
+        )
 
-    for l in languages:
+    for lang in languages:
         out_dict = {}
-        if l == "jpn":
+        if lang == "jpn":
-            for x in result:
+            for key, value in quests.items():
-                out_dict[str(x.questMstId)] = {"Name": x.name, "Enabled": False}
+                out_dict[key] = {"Name": value.name, "Enabled": False}
         else:
-            for x in result:
+            for key, value in quests.items():
-                out_dict[str(x.questMstId)] = {"Name": "", "Enabled": False}
+                out_dict[key] = {"Name": "", "Enabled": False}
 
-        langfile = os.path.join(old_path, "XduQuestNames_{}.json".format(l))
+        langfile = os.path.join(old_path, f"XduQuestNames_{lang}.json")
         if os.path.isfile(langfile):
             with open(langfile, "r") as lang_file:
                 lang_dict = json.load(lang_file)
                 out_dict.update(lang_dict)
 
-        with io.open(
+        with io.open(langfile, "w", newline="\n") as lang_file:
-            langfile, "w", newline="\n"
-        ) as lang_file:  # you're using git right
             json.dump(
-                out_dict, lang_file, ensure_ascii=False, indent="\t", sort_keys=True
+                out_dict,
+                lang_file,
+                ensure_ascii=False,
+                indent="\t",
+                sort_keys=True,
             )
 
     return quests