Fix unused import warning

contrib/affixes.py

@@ -11,12 +11,14 @@ with open(os.path.join(path, "magicprefix.txt")) as f:
     dr = csv.DictReader(f, delimiter="\t")
     index = 0
     for row in dr:
+        if row["Name"] == "Expansion":
+            continue
         index += 1
         if len(row["Name"]) == 0:
             continue
         affixes["prefixes"][index] = {
             "name": row["Name"],
-            "req_lvl": row["levelreq"],
+            "req_lvl": 0 if len(row["levelreq"]) == 0 else int(row["levelreq"]),
             "req_class": None if len(row["class"]) == 0 else row["class"],
         }
 
@@ -24,12 +26,14 @@ with open(os.path.join(path, "magicsuffix.txt")) as f:
     dr = csv.DictReader(f, delimiter="\t")
     index = 0
     for row in dr:
+        if row["Name"] == "Expansion":
+            continue
         index += 1
         if len(row["Name"]) == 0:
             continue
         affixes["suffixes"][index] = {
             "name": row["Name"],
-            "req_lvl": row["levelreq"],
+            "req_lvl": 0 if len(row["levelreq"]) == 0 else int(row["levelreq"]),
             "req_class": None if len(row["class"]) == 0 else row["class"],
         }
 

d2warehouse/app/__init__.py

@@ -1 +1,5 @@
 from d2warehouse.app.main import app
+
+__all__ = [
+    "app",
+]

d2warehouse/app/main.py

@@ -1,7 +1,11 @@
 import hashlib
 from flask import Flask, redirect, abort, render_template, request
 from pathlib import Path
-from d2warehouse.item import Item
+import shutil
+from datetime import datetime
+
+import psutil
+from d2warehouse.item import Item, Quality
 from d2warehouse.parser import parse_stash
 import d2warehouse.db as base_db
 from d2warehouse.app.db import get_db, close_db
@@ -20,6 +24,42 @@ DB_FILES = {
 }
 
 
+def d2_running() -> bool:
+    for proc in psutil.process_iter():
+        try:
+            if proc.cmdline()[0].endswith("D2R.exe"):
+                return True
+        except (IndexError, psutil.AccessDenied):
+            pass
+    return False
+
+
+def storage_count(item: Item, stash: str) -> int | str:
+    """How many of this item type exist in storage"""
+    db = get_stash_db(stash)
+    if item.is_simple:
+        return db.execute(
+            "SELECT COUNT(id) FROM item WHERE code = ?", (item.code,)
+        ).fetchone()[0]
+    elif item.quality == Quality.UNIQUE:
+        return db.execute(
+            "SELECT COUNT(id) FROM item INNER JOIN item_extra ON id = item_id WHERE code = ? AND unique_id = ?",
+            (item.code, item.unique_id),
+        ).fetchone()[0]
+    elif item.quality == Quality.SET:
+        return db.execute(
+            "SELECT COUNT(id) FROM item INNER JOIN item_extra ON id = item_id WHERE code = ? AND set_id = ?",
+            (item.code, item.set_id),
+        ).fetchone()[0]
+    elif item.is_runeword:
+        return db.execute(
+            "SELECT COUNT(id) FROM item INNER JOIN item_extra ON id = item_id WHERE code = ? AND runeword_id = ?",
+            (item.code, item.runeword_id),
+        ).fetchone()[0]
+    else:
+        return "N/A"
+
+
 def save_path() -> Path:
     if "D2SAVE_PATH" in os.environ:
         path = Path(os.environ["D2SAVE_PATH"])
@@ -62,7 +102,11 @@ def list_stash(stash_name: str):
     stash = parse_stash(stash_data)
 
     return render_template(
-        "list_stash.html", stash_name=stash_name, stash=stash, stash_hash=stash_hash
+        "list_stash.html",
+        stash_name=stash_name,
+        stash=stash,
+        stash_hash=stash_hash,
+        storage_count=lambda x: storage_count(x, stash_name),
     )
 
 
@@ -77,6 +121,9 @@ def stash_store_items(stash_name: str):
         return "temp file exists (BAD)"
         return 500
 
+    if d2_running():
+        return "d2 is running", 500
+
     stash_data = stash_path.read_bytes()
     stash_hash = hashlib.sha256(stash_data).hexdigest()
     if request.form.get("stash_hash") != stash_hash:
@@ -94,7 +141,7 @@ def stash_store_items(stash_name: str):
         item = stash.tabs[tab_idx].items[item_idx]
         items.append((tab_idx, item))
 
-    # TODO: create backups
+    backup_stash(stash_name)
 
     for tab_idx, item in items:
         stash.tabs[tab_idx].remove(item)
@@ -125,6 +172,48 @@ def list_storage(stash_name: str):
     )
 
 
+@app.route("/storage/<stash_name>/<category>")
+def list_storage_category(stash_name: str, category: str):
+    if stash_name not in DB_FILES:
+        abort(404)
+
+    db = get_stash_db(stash_name)
+
+    if category == "uniques":
+        q = db.execute(
+            "SELECT id FROM item INNER JOIN item_extra ON id = item_id WHERE deleted IS NULL AND quality = ?",
+            (int(Quality.UNIQUE),),
+        )
+    elif category == "sets":
+        q = db.execute(
+            "SELECT id FROM item INNER JOIN item_extra ON id = item_id WHERE deleted IS NULL AND quality = ?",
+            (int(Quality.SET),),
+        )
+    elif category == "misc":
+        q = db.execute("SELECT id FROM item WHERE deleted IS NULL AND is_simple = TRUE")
+    else:
+        return "Unexpected category", 400
+
+    rows = q.fetchall()
+    items = {}
+    for row in rows:
+        items[row["id"]] = Item.load_from_db(row["id"], db=db)
+
+    return render_template(
+        "list_storage.html", stash_name=stash_name, storage_items=items
+    )
+
+
+def backup_stash(stash_name: str) -> None:
+    stash_path = save_path() / STASH_FILES[stash_name]
+    backup_path = save_path() / "backups"
+    if not backup_path.exists():
+        backup_path.mkdir(parents=True)
+    ts = datetime.now().strftime("%Y-%m-%d_%H.%M.%S.%f")[:-3]
+    backup_path /= f"{ts}_{STASH_FILES[stash_name]}"
+    shutil.copy(stash_path, backup_path)
+
+
 @app.route("/storage/<stash_name>/take", methods=["POST"])
 def storage_take_items(stash_name: str):
     if stash_name not in STASH_FILES or stash_name not in DB_FILES:
@@ -137,14 +226,20 @@ def storage_take_items(stash_name: str):
         return "temp file exists (BAD)"
         return 500
 
+    if d2_running():
+        return "d2 is running", 500
+
     stash_data = stash_path.read_bytes()
     stash = parse_stash(stash_data)
-
-    # TODO: create backups
+    backup_stash(stash_name)
 
     # Write items to temporary stash file
     db = get_stash_db(stash_name)
-    ids = [y.group(1) for x in request.form.keys() if (y := re.match(r"item_(\d+)", x))]
+    ids = [
+        int(y.group(1))
+        for x in request.form.keys()
+        if (y := re.match(r"item_(\d+)", x))
+    ]
     for id in ids:
         item = Item.load_from_db(id, db=db)
         try:

d2warehouse/app/static/style.css

@@ -5,12 +5,48 @@ body {
   color: rgb(240, 240, 240);
 }
 
+.stash-tab {
+  display: grid;
+  grid-template-columns: repeat(5, 1fr);
+  grid-gap: 10px;
+  margin: 0 auto;
+}
+
+@media (max-width: 1600px) {
+  .stash-tab {
+    grid-template-columns: repeat(4, 1fr);
+  }
+}
+
+@media (max-width: 1200px) {
+  .stash-tab {
+    grid-template-columns: repeat(3, 1fr);
+  }
+}
+
+@media (max-width: 800px) {
+  .stash-tab {
+    grid-template-columns: repeat(2, 1fr);
+  }
+}
+
+.stash-tab input[type="checkbox"] {
+  display: none;
+}
+
 .item .name {
   font-weight: bold;
 }
 
 .item {
+  display: block;
   background-color: #444;
+  border: solid #555;
+}
+
+input[type="checkbox"]:checked + label {
+  background-color: #343;
+  border: solid #464;
 }
 
 .color-rare {
@@ -28,3 +64,9 @@ body {
 .color-runeword {
   color: rgb(199, 179, 119);
 }
+
+.raw-item {
+  max-width: 120px;
+  background-color: #444;
+  color: rgb(240, 240, 240);
+}

d2warehouse/app/templates/item.html

@@ -1,14 +1,16 @@
-<div class="item">
-  <input type="checkbox" name="item_{{tabloop.index0}}_{{itemloop.index0}}" value="remove" /> ({{tabloop.index0}}, {{itemloop.index0}})
+<input type="checkbox" id="item_{{tabloop.index0}}_{{itemloop.index0}}" name="item_{{tabloop.index0}}_{{itemloop.index0}}" value="remove" />
+<label class="item" for="item_{{tabloop.index0}}_{{itemloop.index0}}">
   <ul>
     <li class="name color-{{item.color}}">{{item.name}}</li>
     {% if item.quality and item.quality >= 5 %}
     <li class="name color-{{item.color}}">{{item.basename}}</li>
     {% endif %}
+    <li>(in storage: {{storage_count(item)}})</li>
     {% if item.stats %}
     {% for stat in item.stats %}
     <li>{{stat}}</li>
     {% endfor %}
     {% endif %}
+    <li><input class="raw-item" type="text" name="raw item" value="{{item.raw().hex()}}" onfocus="this.select()" readonly></li>
   </ul>
-</div>
+</label>

d2warehouse/app/templates/list_stash.html

@@ -8,14 +8,14 @@
   <body>
     <form action="/stash/{{stash_name}}/store" method="POST">
     {% for tab in stash.tabs %}
-    <div>
       {% set tabloop = loop %}
       <h2>Tab {{tabloop.index}}</h2>
-      {% for item in tab.items %}
-      {% set itemloop = loop %}
-      {% include "item.html" %}
-      {% endfor %}
-    </div>
+      <div class="stash-tab">
+        {% for item in tab.items %}
+        {% set itemloop = loop %}
+        {% include "item.html" %}
+        {% endfor %}
+      </div>
     {% endfor %}
     <input type="submit" value="Store items">
     <input type="hidden" name="stash_hash" value="{{stash_hash}}" />

d2warehouse/item.py

@@ -30,6 +30,7 @@ _stats_map = None
 _unique_map = None
 _set_item_map = None
 _runeword_map = None
+_affix_map = None
 
 
 class Quality(IntEnum):
@@ -282,9 +283,11 @@ class Item:
             # TODO: What affects runeword level? Only the sockets?
             pass
         elif self.quality in [Quality.MAGIC, Quality.RARE, Quality.CRAFTED]:
-            for m, id in [("suffixes", id) for id in self.suffixes] + [
-                ("prefixes", id) for id in self.prefixes
+            for m, id in [(False, id) for id in self.suffixes] + [
+                (True, id) for id in self.prefixes
             ]:
+                if id == 0:
+                    continue
                 affix = lookup_affix(id, m)
                 reqs["lvl"] = max(reqs["lvl"], affix["req_lvl"])
                 if affix["req_class"]:

d2warehouse/parser.py

@@ -268,14 +268,20 @@ def parse_set_data(bits: bitarray, item: Item) -> tuple[Item, int]:
 def parse_rare_data(bits: bitarray, item: Item) -> tuple[Item, int]:
     item.nameword1 = ba2int(bits[0:8])
     item.nameword2 = ba2int(bits[8:16])
-    affixes = []
+    item.prefixes = []
+    item.suffixes = []
     ptr = 16
-    for _ in range(0, 6):
+    for _ in range(0, 3):
+        # Prefix
         (affix, sz) = parse_affix(bits[ptr:])
         ptr += sz
-        affixes.append(affix)
-    item.prefixes = [affix for affix in affixes[0:3] if affix is not None]
-    item.suffixes = [affix for affix in affixes[3:6] if affix is not None]
+        if affix:
+            item.prefixes.append(affix)
+        # Suffix
+        (affix, sz) = parse_affix(bits[ptr:])
+        ptr += sz
+        if affix:
+            item.suffixes.append(affix)
 
     return item, ptr
 

d2warehouse/schema.sql

@@ -32,8 +32,6 @@ CREATE TABLE item (
 );
 -- Add an index for "... WHERE deletion IS NULL"
 CREATE INDEX item_deletion_partial ON item (deleted) WHERE deleted IS NULL;
--- * nuked: if the item has been removed from storage & user indicated he does not
---          want to count it as potentially in his possession any longer
 
 CREATE TABLE item_extra (
     item_id INTEGER PRIMARY KEY,
@@ -65,6 +63,7 @@ CREATE TABLE item_extra (
 
     FOREIGN KEY (item_id) REFERENCES item (id)
 );
+CREATE INDEX item_extra_item_id ON item_extra (item_id);
 
 CREATE TABLE item_stat (
     id INTEGER PRIMARY KEY,
@@ -77,6 +76,7 @@ CREATE TABLE item_stat (
 
     FOREIGN KEY (item_id) REFERENCES item (id)
 );
+CREATE INDEX item_stat_item_id ON item_stat (item_id);
 CREATE INDEX item_stat_stat ON item_stat (stat);
 
 CREATE TABLE item_affix (
@@ -87,3 +87,4 @@ CREATE TABLE item_affix (
 
     FOREIGN KEY (item_id) REFERENCES item (id)
 );
+CREATE INDEX item_affix_item_id ON item_affix (item_id);

d2warehouse/tests/test_parse_item.py

@@ -1,6 +1,6 @@
 import unittest
 from d2warehouse.parser import parse_item
-from d2warehouse.item import Quality, lookup_runeword
+from d2warehouse.item import Quality, lookup_affix, lookup_runeword
 
 
 class ParseItemTest(unittest.TestCase):
@@ -213,3 +213,29 @@ class ParseItemTest(unittest.TestCase):
 
         data, _ = parse_item(data)
         self.assertEqual(data, b"")
+
+    def test_affixes(self):
+        data = bytes.fromhex(
+            "1000800005d0f4aa09173a36bf0723542d351ae4236acbd27000c30201a1052810208cf1241b4c50fc07"
+        )
+        data, item = parse_item(data)
+        self.assertEqual(data, b"")
+
+        req = item.requirements()
+        self.assertEqual(req["lvl"], 4)
+        # +8 stamina
+        self.assertAlmostEqual(lookup_affix(item.prefixes[0], True)["name"], "Rugged")
+        # 16% ed
+        self.assertAlmostEqual(lookup_affix(item.prefixes[1], True)["name"], "Sturdy")
+        # 10% fhr
+        self.assertAlmostEqual(
+            lookup_affix(item.suffixes[0], False)["name"], "of Balance"
+        )
+        # +1 dex
+        self.assertAlmostEqual(
+            lookup_affix(item.suffixes[1], False)["name"], "of Dexterity"
+        )
+        # Lightning bolt on hit
+        self.assertAlmostEqual(
+            lookup_affix(item.suffixes[2], False)["name"], "of Charged Shield"
+        )

pyproject.toml

@@ -19,6 +19,7 @@ license = {text = "GPLv3 License"}
 dependencies = [
     "bitarray",
     "flask",
+    "psutil",
 ]
 dynamic = ["version"]