Half-done collections and UI changes

Minor tweaks and refactors. A half-baked collection support and several UI tweaks.

Just committing these are they are "old" changes and the current state turns out to be pretty stable.
dev
Peery 2 years ago
parent a66c601410
commit b37a21d785

@ -14,12 +14,15 @@ from ArtNet.gui.importer_window import ImporterWindow
from ArtNet.gui.browse_window import BrowseWindow
from ArtNet.gui.dialogs.db_connection_dialog.db_dialog import DBDialog
from ArtNet.web.link_generator import LinkGenerator
from ArtNet.singleton_manager import SingletonManager
class ArtNetManager:
LOG_FOLDER = "log"
def __init__(self, config_location: str = "."):
SingletonManager.set_instance(self)
if not os.path.isdir(ArtNetManager.LOG_FOLDER):
os.mkdir(ArtNetManager.LOG_FOLDER)
file_name = "artnet-" + str(datetime.datetime.now().strftime("%Y-%m-%d")) + ".log"
@ -182,7 +185,7 @@ class ArtNetManager:
if tag in already_applied_tags: # tag is already applied
continue
result = self.db_connection.get_tag_impliers(tag)
result = self.db_connection.get_tag_impliers_by_name(tag)
if len(result) != 0: # tag is implied by some other tag
skip = False
for implier_tag_id in result:
@ -389,10 +392,12 @@ class ArtNetManager:
s = self.all_images[self.curr_image_index].split(os.path.sep)
if image_db_result is not None:
image_title = image_db_result["title"] if isinstance(image_db_result["title"], str) and \
len(image_db_result["title"]) > 0 else self.import_window.UNKNOWN_TITLE
len(image_db_result[
"title"]) > 0 else self.import_window.UNKNOWN_TITLE
image_author = self.db_connection.get_authors_of_art_by_ID(image_db_result["ID"])
image_link = image_db_result["link"]
image_description = image_db_result["description"]
collections = self.db_connection.get_collections_of_art_by_ID(image_db_result["ID"])
art_ID = image_db_result["ID"]
if image_author is None or len(image_author) == 0:
image_author = [(self.all_images[self.curr_image_index].split(os.path.sep)[0], "(Not in Database)")]
@ -401,13 +406,16 @@ class ArtNetManager:
image_author = [(s[0], "(Not in Database)")]
image_title = s[-1] + " (Not in Database)"
image_link = "(Unknown)"
collections = []
image_description = None
logging.info(f"Displaying #{self.curr_image_index} \"{self.all_images[self.curr_image_index]}\"")
self.curr_active_window.display_image(image_title, image_author,
os.path.join(self.config.data["file_root"], self.all_images[self.curr_image_index]),
os.path.join(self.config.data["file_root"],
self.all_images[self.curr_image_index]),
self.all_images[self.curr_image_index],
art_ID, image_link, file_name=s[-1], description=image_description)
art_ID, image_link, file_name=s[-1], description=image_description,
collections=collections)
self.curr_active_window.set_tag_list(
[self.db_connection.get_tag_by_ID(x)[0][1].strip() for x in self.db_connection.get_art_tags_by_ID(art_ID)])
@ -431,7 +439,7 @@ class ArtNetManager:
"""
if len(path) == 0:
exit(0)
logging.info("Changing root to", path)
logging.info(f"Changing root to {path}")
self.config.data["file_root"] = path
self.config.update_config()
self.__file_reader = FileReader(self.config.data["file_root"])
@ -442,6 +450,12 @@ class ArtNetManager:
return self.config.data["db"]
def change_db_connection(self, host: str, port: int, database: str, user: str, password: str):
try:
self.db_connection = self.create_db_connection(host, port, database, user, password)
except ValueError as e:
logging.error(f"Error encountered during change_db_connection(): {e}")
raise e
self.config.data["db"]["user"] = user
self.config.data["db"]["password"] = password
self.config.data["db"]["host"] = host
@ -449,5 +463,3 @@ class ArtNetManager:
self.config.data["db"]["port"] = str(port)
self.config.update_config()
self.db_connection = self.create_db_connection(host, port, database, user, password)

@ -1,10 +1,11 @@
import logging
from typing import List, Tuple
import psycopg2
from psycopg2.errorcodes import UNIQUE_VIOLATION
class DBAdapter:
EXPECTED_TABLES = ['art', 'artist', 'tag', 'topic', 'presence', 'artist_to_topic', 'art_to_presence',
'art_to_tag', 'tag_alias', 'tag_implication', 'tag_category', 'art_collection',
'art_to_art_collection']
@ -30,8 +31,13 @@ class DBAdapter:
Check if all the expected tables are present in the connected database.
:return:
"""
try:
self.db_cursor.execute("select relname from pg_class where relkind='r' and relname !~ '^(pg_|sql_)';")
present_tables = self.db_cursor.fetchall()
except psycopg2.DatabaseError as e:
logging.error(f"Database error occured on check_tables(): {e}")
self.db.rollback()
return False
unknown_tables = []
missing = [x for x in DBAdapter.EXPECTED_TABLES]
for table in present_tables:
@ -46,11 +52,13 @@ class DBAdapter:
return False
if len(unknown_tables) > 0:
logging.error("The following tables are unknown and not expected inside the database: {0}".format(unknown_tables))
logging.error(
"The following tables are unknown and not expected inside the database: {0}".format(unknown_tables))
return True
def save_image(self, ID: str, title: str, authors: list, path: str, tags: list, link: str, md5_hash: str, desc: str):
def save_image(self, ID: str, title: str, authors: list, path: str, tags: list, link: str, md5_hash: str,
desc: str, collections: list):
"""
Updates or saves the given image data to the DB
:param ID:
@ -61,30 +69,34 @@ class DBAdapter:
:param link:
:param md5_hash: md5 hash as a hex digest
:param desc: image description or None for empty
:param collections: list of collections included by the image
:return:
"""
logging.debug("Saving Image {0}:{1} authors: {2} path: {3} tags: {4} link: {5} hash:{6} desc:{7}"
.format(ID, title, authors, path, tags, link, md5_hash, desc))
d = {"title": title, "path": path, "id": ID, "link": link, "hash": md5_hash, "desc": desc}
d = {"title": title, "path": path, "id": ID, "link": link, "hash": md5_hash, "desc": desc} # TODO implement try/except for database errors and rollback!
if self.get_art_by_path(path) is None:
if ID is None:
self.db_cursor.execute(
"INSERT INTO art (path, title, link, md5_hash, description) "
"VALUES (%(path)s, %(title)s, %(link)s, %(hash)s, %(desc)s)",
d)
self.db.commit()
ID = self.get_art_by_path(path)["ID"]
d["id"] = ID
elif self.get_tag_by_ID(ID) is None:
self.db_cursor.execute(
"INSERT INTO art (path, title, link, md5_hash, description) "
"VALUES (%(path)s, %(title)s, %(link)s, %(hash)s, %(desc)s)", d)
self.db.commit()
else:
self.db_cursor.execute("UPDATE art SET path = %(path)s, title = %(title)s, link = %(link)s, " +
"md5_hash = %(hash)s, description = %(desc)s WHERE id = %(id)s", d)
self.db.commit()
if ID is None:
ID = self.get_art_by_path(path)["ID"]
assert(ID != None) # was checked before, should never fail
assert (ID != None) # was checked before, should never fail
old_tags = [self.get_tag_by_ID(x)[0][1].strip() for x in self.get_art_tags_by_ID(ID)]
for tag in tags:
@ -93,10 +105,12 @@ class DBAdapter:
d = {"id": ID, "tag": self.get_tag_ID(tag)}
try:
self.db_cursor.execute("INSERT INTO art_to_tag (art_id, tag_ID) VALUES (%(id)s, %(tag)s)", d)
self.db.commit()
except psycopg2.Error as e:
if e.pgcode == UNIQUE_VIOLATION:
if e.pgcode == UNIQUE_VIOLATION: # tag was already set
logging.debug(e)
logging.info("Skipping Unique Violation ...")
self.db.rollback()
else:
raise e
@ -111,6 +125,7 @@ class DBAdapter:
d = {"id": ID, "author_name": author[0], "author_domain": author[1]}
self.db_cursor.execute("INSERT INTO art_to_presence (presence_name, presence_domain, art_ID) " +
"VALUES (%(author_name)s, %(author_domain)s, %(id)s)", d)
self.db.commit()
for old_author_name, old_author_domain in old_authors:
if (old_author_name, old_author_domain) not in authors: # need to remove old author
self.remove_artist_from_image(art_ID=ID, presence_name=old_author_name,
@ -130,16 +145,15 @@ class DBAdapter:
art_ID = image_data["ID"]
logging.debug(f"Deleting image #{art_ID} {image_data['title']}")
tags = self.get_art_tags_by_ID(art_ID)
for tag_ID in tags:
pass
#self.remove_tag_from_image(art_ID=art_ID, tag_ID=tag_ID)
# for tag_ID in tags:
# pass
# self.remove_tag_from_image(art_ID=art_ID, tag_ID=tag_ID)
authors = self.get_authors_of_art_by_ID(art_ID)
for presence_name, presence_domain in authors:
pass
#self.remove_artist_from_image(art_ID=art_ID, presence_name=presence_name, presence_domain=presence_domain)
#self.db_cursor.execute("DELETE FROM art WHERE md5_hash = %(hash)s", {"hash": hash})
# for presence_name, presence_domain in authors:
# pass
# self.remove_artist_from_image(art_ID=art_ID, presence_name=presence_name, presence_domain=presence_domain)
# self.db_cursor.execute("DELETE FROM art WHERE md5_hash = %(hash)s", {"hash": hash})
def remove_artist_from_image(self, art_ID, presence_name, presence_domain):
"""
@ -340,6 +354,29 @@ class DBAdapter:
"""
self.db_cursor.execute("SELECT id, name FROM artist")
def get_art_by_ID(self, ID: int):
"""
Queries the database for the art via its ID and returns it if available.
"""
d = {"ID": ID}
self.db_cursor.execute("SELECT id, path, title, link, md5_hash, description FROM art WHERE id = %(ID)s", d)
result = self.db_cursor.fetchall()
if len(result) == 0:
return None
else:
result = result[0]
image_data = {
"ID": result[0],
"path": result[1],
"title": result[2],
"link": result[3],
"md5_hash": result[4],
"description": result[5],
}
return image_data
def get_art_by_hash(self, file_hash: str) -> dict:
"""
Queries the database for the art via its hash and returns it if available.
@ -349,7 +386,8 @@ class DBAdapter:
:return:
"""
d = {"hash": file_hash}
self.db_cursor.execute("SELECT id, path, title, link, md5_hash, description FROM art where md5_hash = %(hash)s", d)
self.db_cursor.execute("SELECT id, path, title, link, md5_hash, description FROM art where md5_hash = %(hash)s",
d)
result = self.db_cursor.fetchall()
@ -449,6 +487,36 @@ class DBAdapter:
new_rows.append(row[0])
return new_rows
def search_art(self, tags: list, presences: list, resolve_tags: bool = True):
"""
Search with the tags and given presences for fitting art.
:param tags: List[int] list of tag ids that are to be present in the art. See resolve_tags for implication.
:param presences: List[Tuple(str, str)] list of presences to be present in the art.
:param resolve_tags: flag if tag implications should be resolved fully.
"""
d = {"tags": tags, "presences": presences}
l = list()
for i in range(len(presences)): # converting for psycopg2
name, domain = presences[i]
casted = [name, domain]
l.append(casted)
d["presences"] = l
search_query = "SELECT art.id FROM art " \
"INNER JOIN art_to_tag ON art_to_tag.art_id = art.id " \
"INNER JOIN tag ON art_to_tag.tag_id = tag.id " \
"INNER JOIN art_to_presence as AtP ON art.id = AtP.art_id " \
"GROUP BY art.id " \
"HAVING array_agg(tag.name) @> %(tags)s::varchar[] AND " \
"array_agg(array[AtP.presence_name, AtP.presence_domain]) @> %(presences)s::varchar[]"
# 1. Join all related tables
# 2. Group them over art.id
# 3. check which rows contain the searched tags or presences in the value arrays
self.db_cursor.execute(search_query, d)
result = self.db_cursor.fetchall()
return result
def search_fuzzy_presence(self, name: str, domain: str, all_if_empty: bool = False):
"""
Search a list of presences fitting the (name, domain) fuzzy.
@ -460,7 +528,7 @@ class DBAdapter:
if all_if_empty and (name is None or len(name) == 0) and (domain is None or len(domain) == 0):
self.db_cursor.execute("SELECT name, domain, artist_id FROM presence")
else:
d = {"name":"%"+name+"%", "domain": "%"+domain+"%"}
d = {"name": "%" + name + "%", "domain": "%" + domain + "%"}
self.db_cursor.execute("SELECT name, domain, artist_id FROM presence WHERE LOWER(name) LIKE "
"LOWER(%(name)s) AND domain LIKE %(domain)s", d)
rows = self.db_cursor.fetchall()
@ -502,7 +570,7 @@ class DBAdapter:
:param search:
:return:
"""
d = {"search": "%"+search+"%"}
d = {"search": "%" + search + "%"}
self.db_cursor.execute("SELECT name FROM tag_category WHERE LOWER(name) LIKE LOWER(%(search)s)", d)
rows = []
@ -519,10 +587,11 @@ class DBAdapter:
:return:
"""
if all_if_empty and len(name) == 0:
self.db_cursor.execute("SELECT name, description, category_id FROM tag")
self.db_cursor.execute("SELECT name, description, category_id, ID FROM tag")
else:
d = {"name": "%"+name+"%"}
self.db_cursor.execute("SELECT name, description, category_id FROM tag WHERE LOWER(name) LIKE LOWER(%(name)s)", d)
d = {"name": "%" + name + "%"}
self.db_cursor.execute(
"SELECT name, description, category_id, ID FROM tag WHERE LOWER(name) LIKE LOWER(%(name)s)", d)
rows = self.db_cursor.fetchall()
new_rows = []
@ -553,7 +622,7 @@ class DBAdapter:
elif all_if_empty and ID is None and len(name) == 0:
self.db_cursor.execute("SELECT id, name FROM artist")
else:
d = {"name": "%"+name+"%"}
d = {"name": "%" + name + "%"}
self.db_cursor.execute("SELECT id, name FROM artist WHERE LOWER(name) LIKE LOWER(%(name)s)", d)
return self.db_cursor.fetchall()
@ -600,7 +669,10 @@ class DBAdapter:
"category_id": category_id,
}
self.db_cursor.execute("UPDATE tag SET description = %(description)s, category_id = %(category_id)s, name = %(name)s WHERE tag_ID = %(id)s", d)
self.db_cursor.execute(
"UPDATE tag SET description = %(description)s, category_id = %(category_id)s, name = %(name)s "
"WHERE ID = %(id)s",
d)
if aliases is not None:
old_aliases = self.get_tag_aliases_by_ID(tag_id)
@ -684,7 +756,6 @@ class DBAdapter:
d)
self.db.commit()
def add_implication_by_name(self, name: str, implicant: str):
"""
Add the implication to the database
@ -725,7 +796,7 @@ class DBAdapter:
:return:
"""
d = {"tag": tag}
self.db_cursor.execute("DELETE FROM tag WHERE tag_id = %(tag)s", d)
self.db_cursor.execute("DELETE FROM tag WHERE id = %(tag)s", d)
self.db.commit()
def remove_tag_by_name(self, name: str):
@ -743,11 +814,15 @@ class DBAdapter:
:return:
"""
d = {"name": name}
self.db_cursor.execute("SELECT tag_ID FROM tag where LOWER(name) = LOWER(%(name)s)", d)
try:
self.db_cursor.execute("SELECT ID FROM tag where LOWER(name) = LOWER(%(name)s)", d)
rows = []
for row in self.db_cursor.fetchall():
rows.append(row[0])
except psycopg2.DatabaseError as e:
return None
if len(rows) > 1:
raise Exception("Found multiple tags by the same name!")
@ -765,7 +840,8 @@ class DBAdapter:
:return: Returns the tag's name, description, category, tag ID
"""
d = {"name": name}
self.db_cursor.execute("SELECT name, description, category_id, tag_ID FROM tag where LOWER(name) = LOWER(%(name)s)", d)
self.db_cursor.execute(
"SELECT name, description, category_id, ID FROM tag where LOWER(name) = LOWER(%(name)s)", d)
rows = []
for row in self.db_cursor.fetchall():
@ -789,13 +865,13 @@ class DBAdapter:
:return: Returns the tag's ID, name, description, category
"""
d = {"ID": ID}
self.db_cursor.execute("SELECT tag_ID, name, description, category_id FROM tag where tag_ID = %(ID)s", d)
self.db_cursor.execute("SELECT ID, name, description, category_id FROM tag where ID = %(ID)s", d)
return self.db_cursor.fetchall()
def get_tag_aliases_by_name(self, name: str) -> list:
"""
Search for the tag's aliases and the tag's aliases
Search for the tag's aliases and the alias's aliases
:param name:
:return: List of tag Names that are aliases of this one
"""
@ -803,7 +879,7 @@ class DBAdapter:
result = []
for alias in aliases:
tag_data = self.get_tag_by_ID(alias)
if len(tag_data) > 0:
if len(tag_data) > 0: # tag exists
result.append(tag_data[0][1].strip())
return result
@ -848,6 +924,7 @@ class DBAdapter:
collected_tags.append(found)
to_search.append(found)
if root_ID in collected_tags:
collected_tags.remove(root_ID)
result = []
@ -857,6 +934,31 @@ class DBAdapter:
result.append(tag_data[0][1].strip())
return result
def get_all_tag_impliers_by_ID(self, ID: int) -> list:
"""
Search for all tags that imply this one or imply a tag that implies this one.
:param ID:
:return:
"""
collected_tags = self.get_tag_impliers_by_ID(ID)
to_search = self.get_tag_impliers_by_ID(ID)
collected_tags.append(ID)
while len(to_search) != 0:
curr_tag = to_search.pop()
found_impliers = self.get_tag_impliers_by_ID(curr_tag)
for found in found_impliers:
if found in collected_tags:
continue
else:
collected_tags.append(found)
to_search.append(found)
if ID in collected_tags:
collected_tags.remove(ID)
return collected_tags
def get_tag_implications_by_ID(self, ID: int) -> list:
"""
Search for the tag's implications
@ -908,20 +1010,29 @@ class DBAdapter:
return r
def get_tag_impliers(self, name: str) -> list:
def get_tag_impliers_by_ID(self, ID: int):
"""
Search for tags that imply this one.
:param ID:
:return:
"""
d = {"ID": ID}
self.db_cursor.execute("SELECT root_tag FROM Tag_Implication WHERE implicate = %(ID)s", d)
rows = self.db_cursor.fetchall()
return rows
def get_tag_impliers_by_name(self, name: str) -> list:
"""
Search for tags that imply this one.
:param name:
:return:
"""
d = {"ID": self.get_tag_ID(name)}
self.db_cursor.execute("SELECT root_tag, implicate FROM Tag_Implication WHERE implicate = %(ID)s", d)
self.db_cursor.execute("SELECT root_tag FROM Tag_Implication WHERE implicate = %(ID)s", d)
rows = self.db_cursor.fetchall()
r = []
for row in rows:
r.append(row[0])
return r
return rows
def __get_tag_aliases_no_recurse(self, tag_id: int) -> list:
"""
@ -946,3 +1057,140 @@ class DBAdapter:
raise Exception("Something went terribly wrong!")
return r
def create_collection(self, collection_name: str, description: str) -> int:
"""
Create the collection with name and description as given.
"""
d = {"name": collection_name, "description": description}
try:
self.db_cursor.execute("INSERT INTO art_collection (name, description) VALUES (%(name)s, %(description)s) "
"RETURNING id", d)
self.db.commit()
except psycopg2.DatabaseError as e:
logging.error(f"Encountered database error on create_collection() with collection_name: {collection_name} "
f"description: {description}. Rolling it back now ...")
logging.error(e)
self.db.rollback()
return None
return self.db_cursor.fetchall()[0]
def delete_collection(self, collection_id: int):
"""
Delete the collection given by the id.
"""
d = {"ID": collection_id}
try:
self.db_cursor.execute("DELETE FROM art_collection WHERE ID = %(ID)s", d)
self.db.commit()
except psycopg2.DatabaseError as e:
logging.error(f"Encountered database error on delete_collection() ID: {collection_id}. "
f"Rolling it back now ...")
logging.error(e)
self.db.rollback()
def get_collections_of_art_by_ID(self, art_id: int):
"""
Queries the database for collections that include the given art_id
:return: [(ID, name, description)]
"""
d = {"art_id": art_id}
self.db_cursor.execute("SELECT art_collection.id, art_collection.name, art_collection.description "
"FROM art_collection "
"INNER JOIN art_to_art_collection as atac ON art_collection.id = atac.collection_id "
"WHERE atac.art_id = %(art_id)s", d)
return self.db_cursor.fetchall()
def get_collection(self, collection_id: int):
"""
Returns the collection given by the id.
"""
d = {"ID": collection_id}
self.db_cursor.execute("SELECT collection.id, collection.name, collection.description, array_agg(art.id), "
"array_agg(art.title), array_agg(art.path), array_agg(atac.ranking) "
"FROM art_collection as collection "
"LEFT OUTER JOIN art_to_art_collection as atac ON atac.collection_id = collection.id "
"LEFT OUTER JOIN art ON atac.art_id = art.id "
"WHERE collection.ID = %(ID)s "
"GROUP BY collection.id", d)
r = self.db_cursor.fetchall()
if len(r) > 0:
r = r[0]
collection = dict()
collection["id"] = r[0]
collection["name"] = r[1]
collection["description"] = r[2]
collection["entries"] = []
for i in range(len(r[3])):
art_data = {"id": r[3][i], "title": r[4][i], "path": r[5][i], "ranking": r[6][i]}
if art_data["id"] is None: # skipping non-existing art entries
continue # might happen once when the collection has no entries
collection["entries"].append(art_data)
return collection
def search_collections_by_name(self, name: str) -> List[Tuple[int, str]]:
"""
Returns the collection by the given name
"""
d = {"name": name}
try:
self.db_cursor.execute("SELECT art_collection.id, art_collection.name "
"FROM art_collection "
"WHERE art_collection.name = %(name)s", d)
except psycopg2.DatabaseError as e:
logging.error(f"Encountered database error on get_collection_by_name() name: {name}."
f"Rolling it back now ...")
logging.error(e)
self.db.rollback()
return None
return self.db_cursor.fetchall()
def search_collections(self, collection_name: str = None, presence_name: str = None,
presence_domain: str = None, art_name: str = None):
"""
Search for collections that contain the given parameters.
Search is always fuzzy.
:param collection_name:
:param presence_name:
:param presence_domain:
:param art_name: Search for collections containing art with this title
:result: List[collection_id, collection_name, collection_desc, List[art_id]]
"""
d = {"collection_name": "%"+collection_name+"%", "presence_name": "%"+presence_name+"%",
"presence_domain": "%"+presence_domain+"%", "art_name": "%"+art_name+"%"}
try:
self.db_cursor.execute("SELECT "
"(art_collection.id) as collection_id, "
"(art_collection.name) as collection_name, "
"(art_collection.description) as collection_desc, "
"array_agg(art.id) as art_ID, array_agg(art.title) as art_titles, "
"array_agg(art_to_presence.presence_name) as presence_names, "
"array_agg(art_to_presence.presence_domain) as presence_domains "
"FROM art_collection "
"LEFT OUTER JOIN art_to_art_collection On art_to_art_collection.collection_id = art_collection.id "
"LEFT OUTER JOIN art ON art_to_art_collection.art_id = art.id "
"LEFT OUTER JOIN art_to_presence ON art_to_presence.art_ID = art.id "
"WHERE LOWER(art_collection.name) LIKE LOWER(%(collection_name)s) "
"GROUP BY art_collection.id "
"HAVING LOWER(array_to_string(array_agg(art_to_presence.presence_name), '|', '')) LIKE LOWER(%(presence_name)s) "
"AND LOWER(array_to_string(array_agg(art_to_presence.presence_domain), '|', '')) LIKE LOWER(%(presence_domain)s) "
"AND LOWER(array_to_string(array_agg(art.title), '|', '')) LIKE LOWER(%(art_name)s);",
d)
except psycopg2.DatabaseError as e:
logging.error(f"Encountered database error on search_collections(): {e}")
self.db.rollback()
raise e
return self.db_cursor.fetchall()

@ -32,6 +32,8 @@ class FileReader:
while len(dirs) != 0:
curr_dir = dirs.pop()
curr_full_dir = os.path.join(root, curr_dir)
if not os.path.isdir(curr_full_dir):
continue
l += [os.path.join(curr_dir, f) for f in os.listdir(curr_full_dir) if os.path.isfile(os.path.join(curr_full_dir, f))]
for d in [x for x in os.listdir(curr_full_dir) if os.path.isdir(os.path.join(curr_full_dir, x))]:
dirs.append(os.path.join(curr_dir, d))

@ -41,7 +41,7 @@ class BrowseWindow(ArtnetMainWindow):
self.ui.image_info_button.clicked.connect(self.on_image_info_clicked)
def display_image(self, image_title: str, image_authors: list, full_path: str, relative_path: str, art_ID: int,
link: str, file_name: str, description: str):
link: str, file_name: str, description: str, collections: list):
"""
Display the described image in the GraphicsView
"""

@ -14,7 +14,7 @@ from PyQt5 import QtCore, QtGui, QtWidgets
class Ui_Artist_Mod_Dialog(object):
def setupUi(self, Artist_Mod_Dialog):
Artist_Mod_Dialog.setObjectName("Artist_Mod_Dialog")
Artist_Mod_Dialog.resize(311, 205)
Artist_Mod_Dialog.resize(311, 253)
self.verticalLayout = QtWidgets.QVBoxLayout(Artist_Mod_Dialog)
self.verticalLayout.setObjectName("verticalLayout")
self.dialog_frame = QtWidgets.QFrame(Artist_Mod_Dialog)
@ -49,6 +49,10 @@ class Ui_Artist_Mod_Dialog(object):
sizePolicy.setVerticalStretch(0)
sizePolicy.setHeightForWidth(self.id_label.sizePolicy().hasHeightForWidth())
self.id_label.setSizePolicy(sizePolicy)
font = QtGui.QFont()
font.setBold(True)
font.setWeight(75)
self.id_label.setFont(font)
self.id_label.setObjectName("id_label")
self.verticalLayout_5.addWidget(self.id_label)
self.id_layout = QtWidgets.QVBoxLayout()
@ -65,6 +69,10 @@ class Ui_Artist_Mod_Dialog(object):
sizePolicy.setVerticalStretch(0)
sizePolicy.setHeightForWidth(self.description_label.sizePolicy().hasHeightForWidth())
self.description_label.setSizePolicy(sizePolicy)
font = QtGui.QFont()
font.setBold(True)
font.setWeight(75)
self.description_label.setFont(font)
self.description_label.setObjectName("description_label")
self.verticalLayout_5.addWidget(self.description_label)
self.description_line = QtWidgets.QLineEdit(self.frame)
@ -87,8 +95,8 @@ class Ui_Artist_Mod_Dialog(object):
_translate = QtCore.QCoreApplication.translate
Artist_Mod_Dialog.setWindowTitle(_translate("Artist_Mod_Dialog", "Dialog"))
self.dialog_topic.setText(_translate("Artist_Mod_Dialog", "Artist Topic"))
self.id_label.setText(_translate("Artist_Mod_Dialog", "Artist ID:"))
self.description_label.setText(_translate("Artist_Mod_Dialog", "Artist Description:"))
self.id_label.setText(_translate("Artist_Mod_Dialog", "Artist ID"))
self.description_label.setText(_translate("Artist_Mod_Dialog", "Artist Name"))
if __name__ == "__main__":

@ -7,7 +7,7 @@
<x>0</x>
<y>0</y>
<width>311</width>
<height>205</height>
<height>253</height>
</rect>
</property>
<property name="windowTitle">
@ -63,8 +63,14 @@
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="font">
<font>
<weight>75</weight>
<bold>true</bold>
</font>
</property>
<property name="text">
<string>Artist ID:</string>
<string>Artist ID</string>
</property>
</widget>
</item>
@ -90,8 +96,14 @@
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="font">
<font>
<weight>75</weight>
<bold>true</bold>
</font>
</property>
<property name="text">
<string>Artist Description:</string>
<string>Artist Name</string>
</property>
</widget>
</item>

@ -0,0 +1,247 @@
import logging
import os.path
from PyQt5 import QtWidgets
from PyQt5.QtCore import Qt, QSize, QModelIndex
from PyQt5.QtGui import QStandardItem, QStandardItemModel, QIcon
from ArtNet.gui.dialogs.image_select_dialog.image_select_dialog import Ui_Dialog
from ArtNet.db.db_adapter import DBAdapter
from ArtNet.singleton_manager import SingletonManager
class ImageSelectDialog(QtWidgets.QDialog):
def __init__(self, db_connection, parent=None):
super().__init__(parent)
self.parent = parent
self.__config = SingletonManager.get_manager().config.data
self.__file_root = self.__config["file_root"]
self.db_connection: DBAdapter = db_connection
self.data: dict = None # data to be returned by this dialog to the parent
self.__tag_search_result = list()
self.__presence_search_result = list()
self.__art_list_dict = dict()
self.curr_selected_tags = list()
self.curr_selected_presences = list()
self.setWindowTitle("Select Art")
self.ui = Ui_Dialog()
self.ui.setupUi(self)
self.ui.search_tag_line.textChanged.connect(self.on_tag_search_changed)
self.ui.search_presence_name_line.textChanged.connect(self.on_presence_search_changed)
self.ui.search_presence_domain_line.textChanged.connect(self.on_presence_search_changed)
self.ui.art_selection_list.hide()
self.ui.art_selection_list.doubleClicked.connect(self.on_art_doubleclicked)
self.on_presence_search_changed("")
self.on_tag_search_changed("")
def set_tag_search_result_list(self, tags: list):
"""
Set the tag search result list given the tags from the search
"""
item_model = QStandardItemModel(self.ui.search_tag_list)
self.__tag_search_result = tags
for tag in tags:
item = QStandardItem(tag)
flags = Qt.ItemIsEnabled
item.setData(Qt.Unchecked, Qt.CheckStateRole)
flags |= Qt.ItemIsUserCheckable
if self.curr_selected_tags is not None and tag in self.curr_selected_tags:
# already selected, implied or aliased tags
item.setCheckState(Qt.Checked)
item.setFlags(flags)
item_model.appendRow(item)
item_model.itemChanged.connect(self.on_tag_search_item_changed)
self.ui.search_tag_list.setModel(item_model)
def refresh_selected_tags_list(self):
"""
Refreshes the selected tags list according to self.curr_selected_tags
"""
item_model = QStandardItemModel(self.ui.selected_tag_list)
for tag in self.curr_selected_tags:
item = QStandardItem(tag)
flags = Qt.ItemIsEnabled
item.setData(Qt.Unchecked, Qt.CheckStateRole)
flags |= Qt.ItemIsUserCheckable
item.setCheckState(Qt.Checked)
item.setFlags(flags)
item_model.appendRow(item)
item_model.itemChanged.connect(self.on_tag_selected_item_changed)
self.ui.selected_tag_list.setModel(item_model)
self.refresh_image_selection()
def refresh_selected_presences_list(self):
"""
Refreshes the selected presences list according to self.curr_selected_presences
"""
item_model = QStandardItemModel(self.ui.selected_presence_list)
for name, domain in self.curr_selected_presences:
item = QStandardItem(f"{name}:{domain}")
flags = Qt.ItemIsEnabled
item.setData(Qt.Unchecked, Qt.CheckStateRole)
flags |= Qt.ItemIsUserCheckable
item.setCheckState(Qt.Checked)
item.setFlags(flags)
item_model.appendRow(item)
item_model.itemChanged.connect(self.on_presence_selected_item_changed)
self.ui.selected_presence_list.setModel(item_model)
self.refresh_image_selection()
def refresh_image_selection(self):
"""
Refreshes the images available for selection by using self.curr_selected_presences and self.curr_selected_tags
"""
logging.debug(f"Refreshing info selection with tags {self.curr_selected_tags} "
f"and presences {self.curr_selected_presences}")
arts = self.db_connection.search_art(tags=self.curr_selected_tags, resolve_tags=True,
presences=self.curr_selected_presences)
#print("Result:", arts)
self.ui.art_placeholder_label.hide()
self.ui.art_selection_list.show()
self.ui.art_selection_list.setViewMode(QtWidgets.QListView.IconMode)
self.__art_list_dict = dict()
item_model = QStandardItemModel(self.ui.art_selection_list)
items = list()
if len(arts) > 100:
arts = arts[:100]
for art in arts:
art_data = self.db_connection.get_art_by_ID(art)
full_path = os.path.join(self.__file_root, art_data["path"])
if not os.path.isfile(full_path):
continue
icon = QIcon()
icon.addFile(full_path)
flags = Qt.ItemIsEnabled
item = QStandardItem(
art_data["title"] if art_data["title"] is not None else os.path.basename(art_data["path"]))
item.setData(icon, role=Qt.DecorationRole)
item.setData(art_data["ID"], role=Qt.UserRole) # allows the item to be identified again
presences = self.db_connection.get_authors_of_art_by_ID(art_data["ID"])
p = ""
for presence in presences:
p += f"{presence[0]}:{presence[1]}|"
p = p[:-1]
item.setData(f"{art_data['title']} \nby \n{p} \n\nDescription:\n{art_data['description']}", role=Qt.ToolTipRole) # tooltip for the item
item.setFlags(flags)
items.append(item)
self.__art_list_dict[art_data["ID"]] = art_data
#print("Adding to last row", items)
item_model.appendColumn(items)
self.ui.art_selection_list.setIconSize(QSize(150, 150))
self.ui.art_selection_list.setModel(item_model)
def set_presence_search_result_list(self, presences: list):
"""
Set the presence search result list as given by presences.
"""
item_model = QStandardItemModel(self.ui.search_presence_list)
self.__presence_search_result = presences
for name, domain in presences:
item = QStandardItem(f"{name}:{domain}")
flags = Qt.ItemIsEnabled
item.setData(Qt.Unchecked, Qt.CheckStateRole)
flags |= Qt.ItemIsUserCheckable
item.setFlags(flags)
item_model.appendRow(item)
item_model.itemChanged.connect(self.on_presence_search_item_changed)
self.ui.search_presence_list.setModel(item_model)
def on_art_doubleclicked(self, index: QModelIndex):
art_id = index.model().itemData(index)[Qt.UserRole]
print(f"Double-clicked: {self.__art_list_dict[art_id]}")
self.data = self.__art_list_dict[art_id]
self.done(QtWidgets.QDialog.Accepted)
def on_tag_search_changed(self, v: str):
result = self.db_connection.search_fuzzy_tag(v, all_if_empty=True)
self.set_tag_search_result_list([r[0] for r in result])
def on_presence_search_changed(self, v: str):
presence_name = self.ui.search_presence_name_line.text()
presence_domain = self.ui.search_presence_domain_line.text()
result = self.db_connection.search_fuzzy_presence(name=presence_name, domain=presence_domain, all_if_empty=True)
presences = list()
for name, domain in result:
presences.append((name, domain))
self.set_presence_search_result_list(presences)
def on_tag_search_item_changed(self, item: QStandardItem):
logging.info(f"Tag search item {item.text()} has changed!")
if item.checkState() == Qt.Checked:
self.curr_selected_tags.append(item.text())
if item.checkState() == Qt.Unchecked:
self.curr_selected_tags.remove(item.text())
self.refresh_selected_tags_list()
def on_tag_selected_item_changed(self, item: QStandardItem):
logging.info(f"Tag select item {item.text()} has changed!")
self.curr_selected_tags.remove(item.text())
self.refresh_selected_tags_list()
self.set_tag_search_result_list(self.__tag_search_result)
def on_presence_search_item_changed(self, item: QStandardItem):
logging.info(f"Presence search item {item.text()} has changed!")
name, domain = item.text().split(":")
if item.checkState() == Qt.Checked:
self.curr_selected_presences.append((name, domain))
elif item.checkState() == Qt.Unchecked:
self.curr_selected_presences.remove((name, domain))
self.refresh_selected_presences_list()
def on_presence_selected_item_changed(self, item: QStandardItem):
logging.info(f"Presence select item {item.text()} has changed!")
name, domain = item.text().split(":")
self.curr_selected_presences.remove((name, domain))
self.set_presence_search_result_list(self.__presence_search_result)
self.refresh_selected_presences_list()
def exec_(self) -> dict:
if super(ImageSelectDialog, self).exec_() == QtWidgets.QDialog.Rejected:
return None
if self.data is None:
self.done(QtWidgets.QDialog.Rejected)
return self.data

@ -20,6 +20,8 @@ from ArtNet.gui.dockers.presence.presence_dock import PresenceDocker
from ArtNet.gui.dialogs.category_modify_dialog.category_mod_dialog import CategoryModDialog
from ArtNet.gui.dialogs.tag_import_dialog.tag_imp_dialog import TagImportDialog
from ArtNet.gui.dialogs.link_input_dialog.link_dialog import LinkInputDialog
from ArtNet.gui.dialogs.collection_select_dialog.collection_sel_dialog import CollectionSelectDialog
from ArtNet.gui.dialogs.collection_modify_dialog.collection_dialog import CollectionModDialog
from ArtNet.web.link_generator import LinkGenerator
@ -28,6 +30,7 @@ class ImporterWindow(ArtnetMainWindow):
UNKNOWN_TITLE = "-Title Unknown-"
UNKNOWN_PRESENCE = "(Not in Database)"
UNKNOWN_LINK = "No Source Available"
NO_COLLECTION = "No Collections"
def __init__(self, main):
super().__init__(main)
@ -52,6 +55,7 @@ class ImporterWindow(ArtnetMainWindow):
self.curr_tags: list = list()
self.curr_implied_tags: list = list()
self.curr_tag_aliases: list = list()
self.curr_collections: list = list()
self.__data_changed: bool = False
@ -154,6 +158,7 @@ class ImporterWindow(ArtnetMainWindow):
"link": self.curr_link,
"md5_hash": self.main.get_md5_of_image(self.curr_art_path),
"description": new_description,
"collections": self.curr_collections,
}
for presence in self.curr_presences:
@ -166,10 +171,19 @@ class ImporterWindow(ArtnetMainWindow):
msg.exec_()
return False
for coll_id, coll_name in self.curr_collections:
msg = QtWidgets.QMessageBox()
msg.setWindowTitle("Could not save collection!")
msg.setInformativeText("The saving and proper selection of collections is not supported as of now!")
msg.setIcon(QtWidgets.QMessageBox.Warning)
msg.exec_()
break
self.main.db_connection.save_image(ID=image_data["ID"], title=image_data["title"],
authors=image_data["authors"], path=image_data["path"],
tags=image_data["tags"], link=image_data["link"],
md5_hash=image_data["md5_hash"], desc=image_data["description"])
md5_hash=image_data["md5_hash"], desc=image_data["description"],
collections=image_data["collections"])
self.set_temporary_status_message("Saved {0} ({1}) to ArtNet DB!"
.format(image_data["title"], image_data["ID"]), 5000)
@ -318,6 +332,30 @@ class ImporterWindow(ArtnetMainWindow):
self.data_changed = True
def get_current_collections(self) -> list:
"""
Get the collections currently associated with the current art
"""
return self.curr_collections
def set_current_collections(self, collections: list):
"""
Set the collections associated with the current art
:param collections: List[Tuple[coll_id, coll_name]]
"""
self.curr_collections = collections
if len(collections) == 0:
self.ui.collections_label.setText(ImporterWindow.NO_COLLECTION)
else:
s = ""
for coll_id, coll_name, coll_desc in collections:
s += "<a href=\"{0}\">{1}</a>".format("", f"#{coll_id}:{coll_name}")
# currently not linking to anything
s += "|"
s = s[:-1]
self.ui.collections_label.setText(s)
self.data_changed = True
def get_categories(self, search: str, all_if_empty: bool = False):
"""
Fuzzy Query for categories in the database.
@ -332,6 +370,35 @@ class ImporterWindow(ArtnetMainWindow):
return self.main.db_connection.search_fuzzy_categories(search)
def create_collection(self, collection_name: str, description: str = None) -> int:
"""
Create a new collection with the given name and description
Returns the assigned ID of the collection
"""
return self.main.db_connection.create_collection(collection_name, description)
def get_collection(self, collection_id: int):
"""
Fetch the collection given by the id. If not present None is returned.
"""
return self.main.db_connection.get_collection(collection_id)
def delete_collection(self, collection_id: int):
"""
Delete the collection given by the id from the database.
"""
self.main.db_connection.delete_collection(collection_id)
def search_collections(self, collection_name: str = None, presence_name: str = None,
presence_domain: str = None, art_name: str = None):
"""
Search for collections that fit the given parameters. All parameters are searched fuzzy.
"""
return self.main.db_connection.search_collections(collection_name=collection_name,
presence_name=presence_name, presence_domain=presence_domain,
art_name=art_name)
def get_image_link_from_label(self) -> str:
"""
Gets the image link from the label if it is a valid link.
@ -432,17 +499,18 @@ class ImporterWindow(ArtnetMainWindow):
:param tags:
:return:
"""
tags.sort()
item_model = QStandardItemModel(self.ui.search_result_list)
for tag in tags:
item = QStandardItem(tag)
flags = Qt.ItemIsEnabled
if tag not in self.curr_imply_tags + self.curr_tag_aliases and tag not in self.curr_tags:
if tag not in self.curr_implied_tags:
# new tag and not implied yet
item.setData(Qt.Unchecked, Qt.CheckStateRole)
flags |= Qt.ItemIsUserCheckable
if self.curr_tags is not None and tag in (self.curr_tags + self.curr_imply_tags + self.curr_tag_aliases):
if self.curr_tags is not None and tag in (self.curr_tags + self.curr_implied_tags + self.curr_tag_aliases):
# already selected, implied or aliased tags
item.setCheckState(Qt.Checked)
item.setFlags(flags)
@ -463,21 +531,37 @@ class ImporterWindow(ArtnetMainWindow):
Also updates the tag implication list if no_implication is False
:param tags:
:param set_checked:
:param no_implication: bool indicating if the implication list should also be updated
:param no_implication: not used
:return:
"""
tags.sort()
self.curr_tags = tags
item_model = QStandardItemModel(self.ui.tag_list)
implied_tags = []
for x in self.curr_tags:
# collect all implied tags into a list
implied_tags += self.main.db_connection.get_all_tag_implications_by_name(x)
self.set_implied_list(implied_tags)
for tag in tags:
item = QStandardItem(tag)
item.setFlags(Qt.ItemIsUserCheckable | Qt.ItemIsEnabled)
item.setData(Qt.Unchecked, Qt.CheckStateRole)
item_model.appendRow(item)
flags = Qt.ItemIsEnabled
if tag not in self.curr_implied_tags:
# new tag and not implied yet
item.setData(Qt.Unchecked, Qt.CheckStateRole)
flags |= Qt.ItemIsUserCheckable
if set_checked:
if tag not in self.curr_implied_tags:
item.setCheckState(Qt.Checked)
else:
item.setCheckState(Qt.Unchecked)
item.setFlags(flags)
item_model.appendRow(item)
item_model.itemChanged.connect(self.on_tag_item_changed)
self.ui.tag_list.setModel(item_model)
@ -490,7 +574,7 @@ class ImporterWindow(ArtnetMainWindow):
:param tags:
:return:
"""
self.curr_imply_tags = tags
self.curr_implied_tags = tags
item_model = QStandardItemModel(self.ui.implied_tag_list)
done = []
for tag in tags:
@ -506,7 +590,7 @@ class ImporterWindow(ArtnetMainWindow):
self.data_changed = True
def display_image(self, image_title: str, image_authors: list, full_path: str, relative_path: str, art_ID: int,
link: str, file_name: str, description: str):
link: str, file_name: str, description: str, collections: list):
"""
Display an image in the central widget
:param image_authors:
@ -517,6 +601,7 @@ class ImporterWindow(ArtnetMainWindow):
:param link:
:param file_name:
:param description:
:param collections:
:return:
"""
self.curr_art_id = art_ID
@ -524,6 +609,8 @@ class ImporterWindow(ArtnetMainWindow):
self.curr_image_title = image_title
self.curr_file_name = os.path.basename(full_path)
self.curr_link = link
self.set_current_collections(collections)
self.set_current_presences(image_authors)
file_ending = relative_path.split(".")[-1]
@ -856,6 +943,11 @@ class ImporterWindow(ArtnetMainWindow):
category=tag_data["category"])
self.on_tag_search_change()
implied_tags = [] # refresh implied tags, the edit/addition might have changed smth
for x in self.curr_tags:
# collect all implied tags into a list
implied_tags += self.main.db_connection.get_all_tag_implications_by_name(x)
def on_tag_deletion_clicked(self):
logging.info("Clicked Tag Deletion!")
dialog = TagSelectDialog(self, delete_tag=True)
@ -946,17 +1038,38 @@ class ImporterWindow(ArtnetMainWindow):
def on_tag_search_item_changed(self, item: QStandardItem):
if item.checkState() == Qt.Checked:
self.curr_tags.append(item.text())
aliases = self.main.db_connection.get_tag_aliases_by_name(item.text())
for alias in aliases:
self.curr_tags.append(alias)
implications = self.main.db_connection.get_all_tag_implications_by_name(item.text())
for implication in implications:
self.curr_tags.append(implication)
if item.checkState() == Qt.Unchecked:
if item.text() in self.curr_tags:
self.curr_tags.remove(item.text())
aliases = self.main.db_connection.get_tag_aliases_by_name(item.text())
for alias in aliases:
if alias in self.curr_tags:
self.curr_tags.remove(alias)
implications = self.main.db_connection.get_all_tag_implications_by_name(item.text())
for implication in implications:
self.curr_tags.remove(implication)
else:
return
raise Exception("Something went terribly wrong!")
self.set_tag_list(self.curr_tags)
self.on_tag_search_change()
def on_tag_item_changed(self, item: QStandardItem):
logging.debug("Item {0} has changed!".format(item.text()))
if item.checkState() == Qt.Unchecked:
if item.text() in self.curr_tags:
aliases = self.main.db_connection.get_tag_aliases_by_name(item.text())
if len(aliases) > 0: # tag has aliases, might need to also remove them
for alias in aliases:
self.curr_tags.remove(alias)
self.curr_tags.remove(item.text())
self.set_tag_list(self.curr_tags)
self.on_tag_search_change()
@ -967,7 +1080,7 @@ class ImporterWindow(ArtnetMainWindow):
tags = self.main.db_connection.search_fuzzy_tag(self.ui.tag_search_bar.text(), all_if_empty=True)
result = []
for tag_name, tag_desc, tag_category in tags:
for tag_name, tag_desc, tag_category, tag_id in tags:
result.append(tag_name)
self.set_tag_search_result_list(result)

@ -29,7 +29,7 @@ class ArtnetMainWindow(QtWidgets.QMainWindow):
self.setting_up_data = False
def display_image(self, image_title: str, image_authors: list, full_path: str, relative_path: str, art_ID: int,
link: str, file_name: str, description: str):
link: str, file_name: str, description: str, collections: list):
"""
Display an image in the central widget
:param image_authors:

@ -0,0 +1,13 @@
class SingletonManager:
__instance = None
@staticmethod
def set_instance(instance):
SingletonManager.__instance = instance
@staticmethod
def get_manager():
if SingletonManager.__instance is not None:
return SingletonManager.__instance

2
DB

@ -1 +1 @@
Subproject commit 7a0b0dfc88e46f1ec31842a34c18d408f23f041d
Subproject commit 4d9059b55790d9e758c129fbbf3c706266e999a4

@ -1,13 +1,9 @@
from ArtNet.artnet_manager import ArtNetManager
# TODO fix DB bug
# TODO 1. Open known image
# TODO 2. edit a tag on the current image & save the edit
# TODO 3. attempt to save the current image
# TODO features:
# TODO 2. Add creation/editing/deletion of Topics (Needs Dialogs)
# TODO 3. add changing/adding Topics to Artists (See dockers/topic/artist_topic_docker.py)
# TODO fix bugs:
# TODO 1. import tags on #237 failed to catch all tags
# TODO 2. pressing prev. Unknown from ~#238 returns wrong unknowns! (always #136, #132, #127, then correct)
if __name__ == "__main__":
am = ArtNetManager()

Loading…
Cancel
Save