diff --git a/01-Load-Automated-Archive-into-Mysql.py b/01-Load-Automated-Archive-into-Mysql.py index a33fab0..39ec749 100755 --- a/01-Load-Automated-Archive-into-Mysql.py +++ b/01-Load-Automated-Archive-into-Mysql.py @@ -1,6 +1,8 @@ +#!/usr/bin/env python3 + +from automated_archive import aa from shared_python.Args import Args from shared_python.Sql import Sql -from automated_archive import aa if __name__ == "__main__": args_obj = Args() @@ -10,8 +12,6 @@ # eg: python 01-Load-Automated-Archive-into-Mysql.py -dh localhost -du root -dt dsa -dd temp_python -a AA -f /path/to/ARCHIVE_DB.pl -o . log.info( - 'Loading Automated Archive file "{0}" into database "{1}"'.format( - args.db_input_file, args.temp_db_database - ) + f'Loading Automated Archive file "{args.db_input_file}" into database "{args.temp_db_database}"' ) aa.clean_and_load_data(args, log) diff --git a/02a-Load-Chapters-to-Working-Table.py b/02a-Load-Chapters-to-Working-Table.py index a2bac7a..7e03d84 100755 --- a/02a-Load-Chapters-to-Working-Table.py +++ b/02a-Load-Chapters-to-Working-Table.py @@ -1,4 +1,5 @@ -# encoding: utf-8 +#!/usr/bin/env python3 + from pymysql import cursors from shared_python.Args import Args @@ -8,7 +9,7 @@ # Given an existing final chapter table, this will use the URL field and chapter location to load the chapter contents def __current_table(table_name, db): - query = "SELECT * FROM `{0}`.`{1}`".format(args.output_database, table_name) + query = f"SELECT * FROM `{args.output_database}`.`{table_name}`" dict_cursor = db.cursor(cursors.DictCursor) dict_cursor.execute(query) return dict_cursor.fetchall() @@ -22,5 +23,5 @@ def __current_table(table_name, db): sql = Sql(args, log) chaps = Chapters(args, sql, log) - log.info("Loading chapters from {0}...".format(args.chapters_path)) + log.info(f"Loading chapters from {args.chapters_path}...") chaps.populate_chapters() diff --git a/02b-Extract-Tags-From-Stories.py b/02b-Extract-Tags-From-Stories.py index 0b6fa77..88bd1bb 100755 --- a/02b-Extract-Tags-From-Stories.py +++ b/02b-Extract-Tags-From-Stories.py @@ -1,10 +1,11 @@ +#!/usr/bin/env python3 + import re from shared_python.Args import Args from shared_python.Sql import Sql from shared_python.Tags import Tags - if __name__ == "__main__": """ Only for non-eFiction archives. @@ -18,9 +19,7 @@ sql = Sql(args, log) tags = Tags(args, sql, log) log.info( - "Processing tags from stories and bookmarks table in {0}".format( - args.temp_db_database - ) + f"Processing tags from stories and bookmarks table in {args.temp_db_database}" ) tag_col_list = {} diff --git a/03-Export-Tags-Authors-Stories.py b/03-Export-Tags-Authors-Stories.py index 07776a3..b05e7e6 100755 --- a/03-Export-Tags-Authors-Stories.py +++ b/03-Export-Tags-Authors-Stories.py @@ -1,8 +1,10 @@ +#!/usr/bin/env python3 + import csv -from shared_python.Args import Args -from shared_python.Sql import Sql import html +from shared_python.Args import Args +from shared_python.Sql import Sql from shared_python.Tags import Tags @@ -39,7 +41,7 @@ def write_csv(data, filename, columns): results = tags.distinct_tags(args.temp_db_database) write_csv( results, - "{0}/{1} - tags.csv".format(args.output_folder, args.archive_name), + f"{args.output_folder}/{args.archive_name} - tags.csv", [ cols["id"], cols["original_tag"], @@ -73,28 +75,16 @@ def write_csv(data, filename, columns): results = sql.execute_and_fetchall( args.temp_db_database, - """ - SELECT s.{0} as "Story ID", s.title as "Title", s.summary as "Summary", a.{1} as "Creator", a.email as "Creator Email", + f""" + SELECT s.{story_id} as "Story ID", s.title as "Title", s.summary as "Summary", a.{author_name} as "Creator", a.email as "Creator Email", "" as "New Email address", "" as "AO3 Account? (& does email match?)", "" as "Searched/Found", "" as "Work on AO3?", "" as "Import status", "" as "importer/inviter", "" as "import/invite date", "" as "AO3 link", "" as "Notes (if any)" - FROM {2} ia join {3} a on ia.{4} = a.{5} join {6} s on ia.{7} = s.{8} where ia.item_type = "story"; - """.format( - story_id, - author_name, - item_authors_table, - author_table, - ia_author_col, - author_id, - stories_table, - ia_item_col, - story_id, - ), + FROM {item_authors_table} ia join {author_table} a on ia.{ia_author_col} = a.{author_id} join {stories_table} s on ia.{ia_item_col} = s.{story_id} where ia.item_type = "story"; + """, ) write_csv( results, - "{0}/{1} - authors with stories.csv".format( - args.output_folder, args.archive_name - ), + f"{args.output_folder}/{args.archive_name} - authors with stories.csv", [ "Story ID", "Title", @@ -117,9 +107,9 @@ def write_csv(data, filename, columns): log.debug( f"Exporting authors with bookmarks from {args.temp_db_database} to {args.output_folder}" ) - author_table = "{0}.authors".format(args.temp_db_database) - bookmarks_table = "{0}.story_links".format(args.temp_db_database) - item_authors_table = "{0}.item_authors".format(args.temp_db_database) + author_table = f"{args.temp_db_database}.authors" + bookmarks_table = f"{args.temp_db_database}.story_links" + item_authors_table = f"{args.temp_db_database}.item_authors" author_name = "name" bookmark_id = "id" bookmark_author_col = "author_id" @@ -130,29 +120,17 @@ def write_csv(data, filename, columns): results = sql.execute_and_fetchall( args.temp_db_database, - """ - SELECT s.{0} as "Bookmark ID", s.title as "Title", s.summary as "Summary", a.{1} as "Creator", a.email as "Creator Email", + f""" + SELECT s.{bookmark_id} as "Bookmark ID", s.title as "Title", s.summary as "Summary", a.{author_name} as "Creator", a.email as "Creator Email", s.url as "URL", "" as "New Email address", "" as "AO3 Account? (& does email match?)", "" as "Searched/Found", "" as "Work on AO3?", "" as "Import status", "" as "importer/inviter", "" as "import/invite date", "" as "AO3 link", "" as "Notes (if any)" - FROM {2} ia join {3} a on ia.{4} = a.{5} join {6} s on ia.{7} = s.{8} where ia.item_type = "story_link"; - """.format( - bookmark_id, - author_name, - item_authors_table, - author_table, - ia_author_col, - author_id, - bookmarks_table, - ia_item_col, - bookmark_id, - ), + FROM {item_authors_table} ia join {author_table} a on ia.{ia_author_col} = a.{author_id} join {bookmarks_table} s on ia.{ia_item_col} = s.{bookmark_id} where ia.item_type = "story_link"; + """, ) write_csv( results, - "{0}/{1} - authors with bookmarks.csv".format( - args.output_folder, args.archive_name - ), + f"{args.output_folder}/{args.archive_name} - authors with bookmarks.csv", [ "Bookmark ID", "Title", diff --git a/04-Rename-Tags.py b/04-Rename-Tags.py index 2316ca5..6a1f17b 100755 --- a/04-Rename-Tags.py +++ b/04-Rename-Tags.py @@ -1,4 +1,5 @@ -# encoding: utf-8 +#!/usr/bin/env python3 + import csv from shared_python.Args import Args @@ -6,7 +7,6 @@ from shared_python.Sql import Sql from shared_python.Tags import Tags - if __name__ == "__main__": """ When Tag Wrangling have finished mapping the tags in Google Drive, export the spreadsheet as a CSV file. This script diff --git a/05-Create-Open-Doors-Tables.py b/05-Create-Open-Doors-Tables.py index d0303de..8a05f61 100755 --- a/05-Create-Open-Doors-Tables.py +++ b/05-Create-Open-Doors-Tables.py @@ -1,4 +1,5 @@ -# encoding: utf-8 +#!/usr/bin/env python3 + import os from shared_python.Args import Args @@ -15,7 +16,7 @@ def _clean_email(author): email = author["email"] if email is None or email == "": email = ( - "{0}{1}Archive@ao3.org".format(author["name"], args.archive_name) + "{}{}Archive@ao3.org".format(author["name"], args.archive_name) .replace(" ", "") .replace("'", "") ) @@ -28,7 +29,7 @@ def main(args, log): sql = Sql(args, log) final = FinalTables(args, sql, log) - log.info("Creating final destination tables in {0}".format(args.output_database)) + log.info(f"Creating final destination tables in {args.output_database}") table_names = { "authors": "authors", @@ -50,7 +51,7 @@ def main(args, log): if os.path.exists(args.story_ids_to_remove): with open(args.story_ids_to_remove, "rt") as f: log.info( - "Removing {0} Do Not Import stories...".format( + "Removing {} Do Not Import stories...".format( sum(line.count(",") for line in f) + 1 ) ) @@ -63,7 +64,7 @@ def main(args, log): if args.bookmark_ids_to_remove and os.path.exists(args.bookmark_ids_to_remove): with open(args.bookmark_ids_to_remove, "rt") as f: log.info( - "Removing {0} Do Not Import bookmarks...".format( + "Removing {} Do Not Import bookmarks...".format( sum(line.count(",") for line in f) + 1 ) ) @@ -75,25 +76,19 @@ def main(args, log): stories_without_tags = final.original_table( table_names["stories"], story_exclusion_filter ) - log.info( - "Stories without tags after removing DNI: {0}".format(len(stories_without_tags)) - ) + log.info(f"Stories without tags after removing DNI: {len(stories_without_tags)}") bookmarks_without_tags = final.original_table( table_names["story_links"], bookmark_exclusion_filter ) if bookmarks_without_tags: log.info( - "Bookmarks without tags after removing DNI: {0}".format( - len(bookmarks_without_tags) - ) + f"Bookmarks without tags after removing DNI: {len(bookmarks_without_tags)}" ) else: log.info("No bookmarks to remove") # STORIES - log.info( - "Copying stories to final table {0}.stories...".format(args.output_database) - ) + log.info(f"Copying stories to final table {args.output_database}.stories...") final_stories = [] for story in stories_without_tags: story_authors = final.original_table( @@ -113,9 +108,7 @@ def main(args, log): # BOOKMARKS if bookmarks_without_tags is not None: log.info( - "Copying bookmarks to final table {0}.story_links...".format( - args.output_database - ) + f"Copying bookmarks to final table {args.output_database}.story_links..." ) final_bookmarks = [] for bookmark in bookmarks_without_tags: @@ -132,9 +125,7 @@ def main(args, log): # AUTHORS log.info( - "Copying authors to final table {0}.authors, cleaning emails and removing authors with no works...".format( - args.output_database - ) + f"Copying authors to final table {args.output_database}.authors, cleaning emails and removing authors with no works..." ) final_authors = [] authors = final.original_table(table_names["authors"]) @@ -155,30 +146,24 @@ def main(args, log): if chapters: dest_chapter_table = f"{args.output_database}.{table_names['chapters']}" log.info( - "Copying chapters table {0} from source chapters table...".format( - dest_chapter_table - ) + f"Copying chapters table {dest_chapter_table} from source chapters table..." ) - sql.execute("drop table if exists {0}".format(dest_chapter_table)) + sql.execute(f"drop table if exists {dest_chapter_table}") truncate_and_insert = ( - "create table {0} (unique(id), key(story_id)) select * from {1}.{2}".format( + "create table {} (unique(id), key(story_id)) select * from {}.{}".format( dest_chapter_table, args.temp_db_database, table_names["chapters"] ) ) sql.execute(truncate_and_insert) add_auto_increment = ( - "alter table {0} modify id int not null auto_increment".format( - dest_chapter_table - ) + f"alter table {dest_chapter_table} modify id int not null auto_increment" ) sql.execute(add_auto_increment) else: log.info( - "Creating chapters table {0}.chapters from source stories table...".format( - args.output_database - ) + f"Creating chapters table {args.output_database}.chapters from source stories table..." ) final_chapters = final.dummy_chapters(final_stories) final.insert_into_final("chapters", final_chapters) diff --git a/06-Update-Tags-In-Story-Table.py b/06-Update-Tags-In-Story-Table.py index 5ac5e8c..c490b02 100755 --- a/06-Update-Tags-In-Story-Table.py +++ b/06-Update-Tags-In-Story-Table.py @@ -1,11 +1,11 @@ -# encoding: utf-8 +#!/usr/bin/env python3 + from shared_python.Args import Args from shared_python.FinalTables import FinalTables from shared_python.PopulateTags import PopulateTags from shared_python.Sql import Sql from shared_python.Tags import Tags - if __name__ == "__main__": """ Denormalize tags out of the working tags table into comma-separated lists in the stories or story_link tables diff --git a/08-Check-ODAP-Tables.py b/08-Check-ODAP-Tables.py index 8f6b0fd..7d25f54 100755 --- a/08-Check-ODAP-Tables.py +++ b/08-Check-ODAP-Tables.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import sys + from shared_python.Args import Args from shared_python.Sql import Sql diff --git a/README.md b/README.md old mode 100755 new mode 100644 diff --git a/automated_archive/__init__.py b/automated_archive/__init__.py old mode 100755 new mode 100644 diff --git a/automated_archive/aa.py b/automated_archive/aa.py old mode 100755 new mode 100644 index 6d86166..1823734 --- a/automated_archive/aa.py +++ b/automated_archive/aa.py @@ -1,10 +1,8 @@ -# -- coding: utf-8 -- - -from datetime import datetime import codecs -import re import html +import re import urllib.request +from datetime import datetime from pymysql import connect @@ -27,7 +25,8 @@ def _clean_file(filepath, log): ) if encoding is None or encoding == "": encoding = "utf-8" - archive_db = codecs.open(filepath, "r", encoding=encoding).read() + with codecs.open(filepath, "r", encoding=encoding) as fh: + archive_db = fh.read() # Manually escape single quote entity and reformat file as a Python dictionary step1 = html.unescape(archive_db.replace("'", "\\'")) @@ -58,9 +57,9 @@ def _clean_file(filepath, log): # List fields in AA db file keys = [dict.keys() for dict in archive_db_python.values()] - unique_keys = set([val for sublist in keys for val in sublist]) + unique_keys = {val for sublist in keys for val in sublist} log.info( - "Fields in ARCHIVE_DB.pl: {0}".format(", ".join(str(e) for e in unique_keys)) + "Fields in ARCHIVE_DB.pl: {}".format(", ".join(str(e) for e in unique_keys)) ) return archive_db_python @@ -142,10 +141,8 @@ def _extract_date(args, record, log): dt = datetime.fromtimestamp(int(date_string)) else: dt = datetime.strptime(date_string, "%m/%d/%y") - except Exception as e: - log.error( - f"Failed to parse date value '{date_string}' due to exception: {str(e)}" - ) + except Exception as e: # noqa: BLE001 + log.error(f"Failed to parse date value '{date_string}' due to exception: {e!s}") return dt.strftime("%Y-%m-%d") if dt else "" @@ -156,9 +153,9 @@ def _create_mysql(args, FILES, log): DATABASE_NAME = args.temp_db_database # Use the database and empty all the tables - cursor.execute("drop database if exists {0};".format(DATABASE_NAME)) - cursor.execute("create database {0};".format(DATABASE_NAME)) - cursor.execute("use {0}".format(DATABASE_NAME)) + cursor.execute(f"drop database if exists {DATABASE_NAME};") + cursor.execute(f"create database {DATABASE_NAME};") + cursor.execute(f"use {DATABASE_NAME}") # Instead of duplicating this file in the repo grab it from the master branch of eFiction url = "https://raw.githubusercontent.com/otwcode/open-doors-eFiction/refs/heads/master/opendoors/open-doors-tables-working.sql" @@ -268,8 +265,8 @@ def _create_mysql(args, FILES, log): } stor = """ - INSERT INTO {0} (id, fandoms, title, summary, tags, characters, date, url, notes, relationships, rating, warnings) - VALUES({1}, '{2}', '{3}', '{4}', '{5}', '{6}', '{7}', '{8}', '{9}', '{10}', '{11}', '{12}');\n""".format( + INSERT INTO {} (id, fandoms, title, summary, tags, characters, date, url, notes, relationships, rating, warnings) + VALUES({}, '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}');\n""".format( table_name, original_id, final_fandoms.replace(r"\\", "\\"), @@ -287,24 +284,9 @@ def _create_mysql(args, FILES, log): cursor.execute(stor) except: log.error( - "table name: {0}\noriginal id: {1}\nfinal fandoms: '{2}'\ntitle: '{3}'\nsummary: '{4}'\ntags: '{5}'" - "\ncharacters: '{6}'\ndate: '{7}'\nfilename: '{8}'\nnotes: '{9}'\npairings: '{10}'\nrating: '{11}'" - "\nwarnings: '{12}'\nauthor id: '{13}'".format( - table_name, - original_id, - final_fandoms, - title, - summary, - tags, - characters, - date, - filename, - notes, - pairings, - rating, - warnings, - authorid, - ) + f"table name: {table_name}\noriginal id: {original_id}\nfinal fandoms: '{final_fandoms}'\ntitle: '{title}'\nsummary: '{summary}'\ntags: '{tags}'" + f"\ncharacters: '{characters}'\ndate: '{date}'\nfilename: '{filename}'\nnotes: '{notes}'\npairings: '{pairings}'\nrating: '{rating}'" + f"\nwarnings: '{warnings}'\nauthor id: '{authorid}'" ) raise db.commit() @@ -313,7 +295,7 @@ def _create_mysql(args, FILES, log): try: item_auth = """ INSERT INTO item_authors (author_id, item_id, item_type) - VALUES({0}, {1}, '{2}');\n""".format( + VALUES({}, {}, '{}');\n""".format( item_info["authorid"], itemid, item_info["itemtype"] ) cursor.execute(item_auth) @@ -331,5 +313,5 @@ def clean_and_load_data(args, log): if __name__ == "__main__": - args = Args().process_args() + args = Args.process_args() data = _clean_file(args.filepath) diff --git a/example.yml b/example.yml old mode 100755 new mode 100644 diff --git a/requirements.txt b/requirements.txt index 476d90d..1b11caf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -17,4 +17,4 @@ pyxdg==0.28 pyyaml==6.0.1 colorlog==6.7.0 prompt_toolkit==3.0.39 -ruff==0.1.5 +ruff diff --git a/ruff.toml b/ruff.toml new file mode 100644 index 0000000..a189712 --- /dev/null +++ b/ruff.toml @@ -0,0 +1,2 @@ +[lint] +ignore = ["N999"] diff --git a/shared_python/Args.py b/shared_python/Args.py old mode 100755 new mode 100644 index 2fa2b8f..c041e71 --- a/shared_python/Args.py +++ b/shared_python/Args.py @@ -1,11 +1,12 @@ -import os import argparse +import os + import yaml from shared_python.Logging import logger -class Args(object): +class Args: def __init__(self): self.args = self._process_args() self.log = logger(self.args.archive_name) @@ -142,7 +143,7 @@ def _process_args(self): else: setattr(args, k, v) - for arg_name in argdict.keys(): + for arg_name in argdict: # noqa: PLC0206 if getattr(args, arg_name) is None: setattr(args, arg_name, input(argdict[arg_name] + ": ")) @@ -157,7 +158,7 @@ def _process_args(self): def _print_args(self, args): self.log.info("----------- Open Door Archive Import Parameters --------------") for arg in vars(args): - self.log.info("{0} = {1}".format(arg, getattr(args, arg))) + self.log.info(f"{arg} = {getattr(args, arg)}") self.log.info("--------------------------------------------------------------") def args_for_01(self): diff --git a/shared_python/Chapters.py b/shared_python/Chapters.py old mode 100755 new mode 100644 index 6d79b35..f4d98c9 --- a/shared_python/Chapters.py +++ b/shared_python/Chapters.py @@ -1,16 +1,15 @@ -# -- coding: utf-8 -- - import codecs import os import re from pip._vendor.distlib.compat import raw_input +from pymysql import MySQLError from shared_python import Common # TODO this code is no longer needed for eFiction and will need to be reviewed for other archive types -class Chapters(object): +class Chapters: def __init__(self, args, sql, log): self.args = args self.sql = sql @@ -38,16 +37,14 @@ def _gather_and_dedupe(self, chapters_path, extensions, has_ids=False): if has_ids and self._ends_with(filename, extensions): file_path = os.path.join(root, filename) cid = os.path.splitext(filename)[0] - if cid not in file_paths.keys(): + if cid not in file_paths: file_paths[cid] = file_path else: duplicate_folder = os.path.split(os.path.split(file_path)[0])[1] messages.append( file_path + " is a duplicate of " + file_paths[cid] ) - sql_messages.append( - "SELECT * FROM chapters WHERE id = {0}".format(cid) - ) + sql_messages.append(f"SELECT * FROM chapters WHERE id = {cid}") duplicate_chapters[cid] = [ { "folder_name": os.path.split( @@ -69,30 +66,28 @@ def _gather_and_dedupe(self, chapters_path, extensions, has_ids=False): file_paths[name] = file_path if has_duplicates: - self.log.warn("\n".join(messages + sql_messages)) - self.log.warn(duplicate_chapters) + self.log.warning("\n".join(messages + sql_messages)) + self.log.warning(duplicate_chapters) folder_name_type = raw_input( - "Resolving duplicates: pick the type of the folder name under {0} " - "\n1 = author id\n2 = author name\n3 = skip duplicates check\n".format( - chapters_path - ) + f"Resolving duplicates: pick the type of the folder name under {chapters_path} " + "\n1 = author id\n2 = author name\n3 = skip duplicates check\n" ) if folder_name_type == "1": for cid, duplicate in duplicate_chapters.items(): # look up the author id and add that one to the file_names list sql_author_id = self.sql.execute_and_fetchall( self.sql.database, - "SELECT author_id FROM chapters WHERE id = {0}".format(cid), + ("""SELECT author_id FROM chapters WHERE id = %s""", [cid]), ) if len(sql_author_id) > 0: author_id = sql_author_id[0][0] - file_paths[cid] = [ + file_paths[cid] = next( dc["path"] - for dc in duplicate_chapters[cid] + for dc in duplicate if dc["folder_name"] == str(author_id) - ][0] + ) elif folder_name_type == "2": - self.log.warn("Not implemented") + self.log.warning("Not implemented") return file_paths @@ -106,7 +101,7 @@ def populate_chapters(self, folder=None, extensions=None): self.log.info("Processing chapters...") filenames_are_ids = raw_input("\nChapter file names are chapter ids? Y/N\n") - has_ids = True if str.lower(filenames_are_ids) == "y" else False + has_ids = str.lower(filenames_are_ids) == "y" file_paths = self._gather_and_dedupe(folder, extensions, has_ids) char_encoding = raw_input( @@ -128,20 +123,16 @@ def populate_chapters(self, folder=None, extensions=None): try: cur = Common.print_progress(cur, total) file_contents = c.read() - query = "UPDATE {0}.chapters SET text=%s WHERE id=%s".format( - self.args.output_database - ) + query = f"UPDATE {self.args.output_database}.chapters SET text=%s WHERE id=%s" self.sql.execute(query, (file_contents, int(cid))) - except Exception as e: + except MySQLError as e: self.log.error( - "Error = chapter id: {0} - chapter: {1}\n{2}".format( - cid, chapter_path, str(e) - ) + f"Error = chapter id: {cid} - chapter: {chapter_path}\n{e!s}" ) finally: pass else: - for _, chapter_path in file_paths.items(): + for chapter_path in file_paths.values(): path = chapter_path.replace(self.args.chapters_path, "")[1:] if os.sep == "\\": # if this script is run on windows path = path.replace("\\", "/") @@ -149,15 +140,11 @@ def populate_chapters(self, folder=None, extensions=None): try: cur = Common.print_progress(cur, total) file_contents = c.read() - query = "UPDATE {0}.chapters SET text=%s WHERE url=%s and text=''".format( - self.args.output_database - ) + query = f"UPDATE {self.args.output_database}.chapters SET text=%s WHERE url=%s and text=''" self.sql.execute(query, (file_contents, path)) - except Exception as e: + except MySQLError as e: self.log.error( - "Error = chapter id: {0} - chapter: {1}\n{2}".format( - path, chapter_path, str(e) - ) + f"Error = chapter id: {path} - chapter: {chapter_path}\n{e!s}" ) finally: pass diff --git a/shared_python/Common.py b/shared_python/Common.py old mode 100755 new mode 100644 index 1214162..728b2e2 --- a/shared_python/Common.py +++ b/shared_python/Common.py @@ -1,4 +1,3 @@ -# -- coding: utf-8 -- import sys from importlib import reload @@ -10,6 +9,6 @@ def print_progress(cur, total, prog_type="stories"): cur += 1 import sys - sys.stdout.write("\r{0}/{1} {2}".format(cur, total, prog_type)) + sys.stdout.write(f"\r{cur}/{total} {prog_type}") sys.stdout.flush() return cur diff --git a/shared_python/FinalTables.py b/shared_python/FinalTables.py old mode 100755 new mode 100644 index 665d4c7..37beabb --- a/shared_python/FinalTables.py +++ b/shared_python/FinalTables.py @@ -1,10 +1,11 @@ import datetime import html from logging import Logger + from shared_python.Sql import Sql -class FinalTables(object): +class FinalTables: def __init__(self, args, sql: Sql, log: Logger): self.args = args self.sql = sql @@ -19,9 +20,7 @@ def original_table(self, table_name, filter="", database_name=None): original_database = self.original_database else: original_database = database_name - query = "SELECT * FROM `{0}`.`{1}` {2}".format( - original_database, table_name, filter - ) + query = f"SELECT * FROM `{original_database}`.`{table_name}` {filter}" return self.sql.execute_dict(query) def _escape_unescape(self, item): @@ -53,9 +52,7 @@ def insert_into_final(self, output_table_name, rows, target_database=None): final_database = target_database else: final_database = self.final_database - self.sql.execute( - "TRUNCATE `{0}`.`{1}`".format(final_database, output_table_name) - ) + self.sql.execute(f"TRUNCATE `{final_database}`.`{output_table_name}`") columns = rows[0].keys() values = [] for row in rows: @@ -64,7 +61,7 @@ def insert_into_final(self, output_table_name, rows, target_database=None): self.sql.execute( f""" - INSERT INTO `{final_database}`.`{output_table_name}` ({', '.join(columns)}) VALUES {', '.join(values)} + INSERT INTO `{final_database}`.`{output_table_name}` ({", ".join(columns)}) VALUES {", ".join(values)} """ ) @@ -72,13 +69,13 @@ def populate_story_tags(self, story_id, output_table_name, story_tags): cols_with_tags = [] for col, tags in story_tags.items(): cols_with_tags.append( - "{0}='{1}'".format(col, tags.replace("'", "\\'").strip()) + "{}='{}'".format(col, tags.replace("'", "\\'").strip()) ) if cols_with_tags: self.sql.execute( """ - UPDATE `{0}`.`{1}` SET {2} WHERE id={3} + UPDATE `{}`.`{}` SET {} WHERE id={} """.format( self.final_database, output_table_name, diff --git a/shared_python/Logging.py b/shared_python/Logging.py index 6bad80e..2c8414e 100644 --- a/shared_python/Logging.py +++ b/shared_python/Logging.py @@ -1,5 +1,6 @@ import logging import sys + from colorlog import ColoredFormatter @@ -14,7 +15,7 @@ def logger(filename): log.addHandler(stream) formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s") - fh = logging.FileHandler("{0}.log".format(filename)) + fh = logging.FileHandler(f"{filename}.log") fh.setFormatter(formatter) log.addHandler(fh) return log diff --git a/shared_python/PopulateTags.py b/shared_python/PopulateTags.py index f69c012..1509bd1 100644 --- a/shared_python/PopulateTags.py +++ b/shared_python/PopulateTags.py @@ -3,7 +3,7 @@ CNTW = "Choose Not To Use Archive Warnings" -class PopulateTags(object): +class PopulateTags: def __init__(self, args, db, log, tags, final): self.args = args self.db = db @@ -26,10 +26,8 @@ def tags_for_story(self, story_id, tags_by_type): fandoms = [] for tag_type, tag_type_tags in tags_by_type.items(): if tag_type is None or tag_type == "": - self.log.warn( - "\nStory {2} has a None tag type\n {0} -> {1}".format( - tag_type, tag_type_tags, story_id - ) + self.log.warning( + f"\nStory {story_id} has a None tag type\n {tag_type} -> {tag_type_tags}" ) else: tag_list = [ @@ -47,7 +45,7 @@ def tags_for_story(self, story_id, tags_by_type): fandoms = [self.args.default_fandom] story_tags["categories"] = ", ".join(set(categories)) story_tags["fandoms"] = ", ".join(set(fandoms)) - if "warnings" not in story_tags.keys(): + if "warnings" not in story_tags: story_tags["warnings"] = CNTW return story_tags @@ -61,8 +59,8 @@ def write_tags_for_story(self, tags_by_story_id, item_type="story"): tags_by_type[tag["ao3_tag_type"]].append(tag) story_tags = self.tags_for_story(story_id, tags_by_type) - self.final.populate_story_tags(story_id, output_table, story_tags) + # is there a method for this one? def populate_tags(self): self.log.info("Getting all tags per story or story_link...") diff --git a/shared_python/Sql.py b/shared_python/Sql.py old mode 100755 new mode 100644 index 54057aa..cdeafd3 --- a/shared_python/Sql.py +++ b/shared_python/Sql.py @@ -2,19 +2,17 @@ import warnings # ignore unhelpful MySQL warnings -from pymysql import connect, cursors, OperationalError +from pymysql import OperationalError, connect, cursors warnings.filterwarnings("ignore", category=Warning) -class Sql(object): +class Sql: def __init__(self, args, log): self.tag_count = 0 conn = connect(host=args.db_host, user=args.db_user, password=args.db_password) cursor = conn.cursor() - cursor.execute( - "CREATE DATABASE IF NOT EXISTS `{0}`".format(args.temp_db_database) - ) + cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{args.temp_db_database}`") self.log = log self.conn = connect( @@ -55,9 +53,8 @@ def execute_and_fetchall(self, database: str, statement: str): def run_script_from_file(self, filename, database, initial_load=False): # Open and read the file as a single buffer - fd = open(filename, "r") - sqlFile = fd.read() - fd.close() + with open(filename, "r") as fd: + sqlFile = fd.read() self.run_sql_file(sqlFile, database, initial_load) def run_sql_file(self, sqlFile, database, initial_load=False): @@ -66,8 +63,8 @@ def run_sql_file(self, sqlFile, database, initial_load=False): # Start a transaction self.cursor.execute("START TRANSACTION") - self.cursor.execute("CREATE DATABASE IF NOT EXISTS {0}".format(database)) - self.cursor.execute("USE {0}".format(database)) + self.cursor.execute(f"CREATE DATABASE IF NOT EXISTS {database}") + self.cursor.execute(f"USE {database}") # Execute every command from the input file for command in sqlCommands: @@ -79,24 +76,23 @@ def run_sql_file(self, sqlFile, database, initial_load=False): end_command = re.sub(r"(--|#|\/\*).*?\n", "", command) lc_command = end_command.lower().strip().replace("\n", "") if initial_load and ( - lc_command.startswith("create database ") - or lc_command.startswith("use ") + lc_command.startswith(("create database ", "use")) ): - self.log.info("Skipping command - {0}".format(lc_command)) + self.log.info(f"Skipping command - {lc_command}") elif lc_command is None or lc_command == "": self.log.info(lc_command) else: self.cursor.execute(lc_command) except OperationalError as e: - self.log.info("Command skipped: {0} [{1}]".format(command, e)) + self.log.info(f"Command skipped: {command} [{e}]") self.conn.commit() def col_exists(self, col, table, database): self.cursor.execute( - """ - SELECT * FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '{0}' AND TABLE_NAME = '{1}' AND COLUMN_NAME = '{2}' - """.format(database, table, col) + f""" + SELECT * FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '{database}' AND TABLE_NAME = '{table}' AND COLUMN_NAME = '{col}' + """ ) result = self.cursor.fetchone() return result is not None diff --git a/shared_python/TagValidator.py b/shared_python/TagValidator.py index 12b2ac2..78793ac 100644 --- a/shared_python/TagValidator.py +++ b/shared_python/TagValidator.py @@ -1,34 +1,39 @@ -class TagValidator(object): +class TagValidator: def __init__(self, log): self.log = log - - ## Dictionaries/constants for tag validation - IS_TAGTYPE = True - dict_tag_type = { - "rating": 1, - "warnings": 2, - "categories": 3, - "fandoms": 4, - "characters": 5, - "relationships": 6, - "tags": 7, - } - dict_rating = { - "Not Rated": 1, - "General Audiences": 2, - "Teen And Up Audiences": 3, - "Mature": 4, - "Explicit": 5, - } - dict_warning = { - "Choose Not To Use Archive Warnings": 1, - "Graphic Depictions Of Violence": 2, - "Major Character Death": 3, - "No Archive Warnings Apply": 4, - "Rape/Non-Con": 5, - "Underage": 6, - } - dict_category = {"Gen": 1, "F/M": 2, "M/M": 3, "F/F": 4, "Multi": 5, "Other": 6} + self.dict_tag_type = { + "rating": 1, + "warnings": 2, + "categories": 3, + "fandoms": 4, + "characters": 5, + "relationships": 6, + "tags": 7, + } + self.IS_TAGTYPE = True + self.dict_rating = { + "Not Rated": 1, + "General Audiences": 2, + "Teen And Up Audiences": 3, + "Mature": 4, + "Explicit": 5, + } + self.dict_warning = { + "Choose Not To Use Archive Warnings": 1, + "Graphic Depictions Of Violence": 2, + "Major Character Death": 3, + "No Archive Warnings Apply": 4, + "Rape/Non-Con": 5, + "Underage": 6, + } + self.dict_category = { + "Gen": 1, + "F/M": 2, + "M/M": 3, + "F/F": 4, + "Multi": 5, + "Other": 6, + } ## Functions for accessing dictionaries def identify_tag_type(self, tag_type): diff --git a/shared_python/Tags.py b/shared_python/Tags.py old mode 100755 new mode 100644 index 6f680fb..a07e219 --- a/shared_python/Tags.py +++ b/shared_python/Tags.py @@ -1,16 +1,16 @@ import re +import sys from collections import defaultdict from html.parser import HTMLParser from logging import Logger -import sys from pymysql import OperationalError from shared_python.Sql import Sql from shared_python.TagValidator import TagValidator -class Tags(object): +class Tags: def __init__(self, args, sql: Sql, log: Logger): self.tag_count = 0 self.sql = sql @@ -36,12 +36,12 @@ def create_tags_table(self, database=None): """ try: database = self.database if database is None else database - self.sql.execute("DROP TABLE IF EXISTS {0}.`tags`".format(database)) + query = """DROP TABLE IF EXISTS %s""" + self.sql.execute(query, [database]) except OperationalError as e: - self.log.info("Command skipped: {}".format(e)) - self.sql.execute( - """ - CREATE TABLE IF NOT EXISTS {0}.`tags` ( + self.log.info(f"Command skipped: {e}") + create_tags_query = """ + CREATE TABLE IF NOT EXISTS %s.`tags` ( `id` int(11) AUTO_INCREMENT, `original_tagid` int(11) DEFAULT NULL, `original_tag` varchar(1024) DEFAULT NULL, @@ -53,9 +53,8 @@ def create_tags_table(self, database=None): `ao3_tag_category` VARCHAR(255) DEFAULT NULL, `ao3_tag_fandom` VARCHAR(255) DEFAULT NULL, PRIMARY KEY (`id`) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - """.format(database) - ) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;""" + self.sql.execute(create_tags_query, [database]) def populate_tag_table( self, @@ -69,19 +68,20 @@ def populate_tag_table( """ Used only in step 02 for non-eFiction archives """ - self.sql.execute("USE {0}".format(database_name)) + use_query = """USE %s""" + truncate_query = """TRUNCATE %s""" + self.sql.execute(use_query, [database_name]) if truncate: - self.sql.execute("TRUNCATE {0}.`tags`".format(database_name)) + self.sql.execute(truncate_query, [database_name]) tag_columns = ( tag_col_lookup.keys() ) # [d['col'] for d in tag_col_lookup if 'col' in d] # Get all values from all tag columns in the stories table and load as denormalised values in `tags` table + select_query = """SELECT %s, %s FROM %s""" data = self.sql.execute_dict( - "SELECT {0}, {1} FROM {2}".format( - story_id_col_name, ", ".join(tag_columns), table_name - ) + select_query, [story_id_col_name, ", ".join(tag_columns), table_name] ) tags_to_insert = {} @@ -90,46 +90,40 @@ def populate_tag_table( for col in tag_columns: needs_fandom = col in tags_with_fandoms if story_tags_row[col] is not None: - for val in re.split(r", ?", story_tags_row[col]): - if val != "": - if isinstance( - tag_col_lookup[col], str - ): # Probably AA or a custom archive - cleaned_tag = re.sub( - r'(? 0: + # change this, leaving alone for the time being self.sql.execute( """ - INSERT INTO tags (original_tag, original_type, ao3_tag_fandom) VALUES {0} + INSERT INTO tags (original_tag, original_type, ao3_tag_fandom) VALUES {} """.format(", ".join(tags_to_insert.values())) ) - + # TODO: change this here, it seems like it just formats the string with the tuples which seems like it could cause problems tag_data = self.sql.execute_dict("SELECT id, original_tag FROM tags") for tag_row in tag_data: story_ids = set(tags_to_story_ids[tag_row["original_tag"]]) for story_id in story_ids: self.sql.execute( - """ - INSERT INTO item_tags (item_id, item_type, tag_id) VALUES ({0}, "{1}", {2}) - """.format( + ( + """ + INSERT INTO item_tags (item_id, item_type, tag_id) VALUES (%s, %s, %s) + """ + ), + [ story_id, "story_link" if table_name == "story_links" else "story", tag_row["id"], - ) + ], ) def distinct_tags(self, database): @@ -169,7 +163,7 @@ def update_tag_row(self, row: dict): :return: number of newly inserted rows to item_tags """ tag_headers = self.tag_export_map - tag = str(row[tag_headers["original_tag"]]).replace("'", r"\'") + tag = str(row[tag_headers["original_tag"]]) tag_id = row[tag_headers["id"]] validate_tag = TagValidator(self.log) @@ -178,8 +172,8 @@ def update_tag_row(self, row: dict): else: tagid_filter = f"id={tag_id}" - fandom = row[tag_headers["ao3_tag_fandom"]].replace("'", r"\'") - ao3_tags = row[tag_headers["ao3_tag"]].replace("'", r"\'").split(",") + fandom = row[tag_headers["ao3_tag_fandom"]] + ao3_tags = row[tag_headers["ao3_tag"]].split(",") ao3_tag_types = row[tag_headers["ao3_tag_type"]].split(",") number_types = len(ao3_tag_types) @@ -202,60 +196,70 @@ def update_tag_row(self, row: dict): if idx > 0: self.sql.execute( - f""" + """ INSERT INTO tags (ao3_tag, ao3_tag_type, ao3_tag_category, ao3_tag_fandom, original_tag, original_tagid) - VALUES ('{ao3_tag}', '{ao3_tag_type}', '{row[tag_headers['ao3_tag_category']]}', - '{fandom}', '{tag}', '{tag_id}') - """ + VALUES (%s, %s, %s, + %s, %s, %s) + """, + [ + ao3_tag, + ao3_tag_type, + row[tag_headers["ao3_tag_category"]], + fandom, + tag, + tag_id, + ], ) # get last auto increment tag id sql_dict = self.sql.execute_dict("""select LAST_INSERT_ID();""") new_tag_id = sql_dict[0]["LAST_INSERT_ID()"] # get all associated items from item_tags items = self.sql.execute_dict( - f"""SELECT item_id, item_type - FROM item_tags WHERE tag_id = {row['Original Tag ID']}""" + """SELECT item_id, item_type + FROM item_tags WHERE tag_id = %s""", + [row["Original Tag ID"]], ) # insert into item_tags table for item in items: item_id, item_type = item["item_id"], item["item_type"] self.sql.execute( - f"""INSERT INTO item_tags (item_id, item_type, tag_id) VALUES ('{item_id}', '{item_type}', '{new_tag_id}')""" + """INSERT INTO item_tags (item_id, item_type, tag_id) VALUES (%s, %s, %s)""", + [item_id, item_type, new_tag_id], ) num_insert += 1 else: + query = f"""UPDATE tags SET ao3_tag=%s, ao3_tag_type=%s, ao3_tag_category=%s, ao3_tag_fandom=%s WHERE {tagid_filter}""" self.sql.execute( - f""" - UPDATE tags - SET ao3_tag='{str(ao3_tag)}', ao3_tag_type='{ao3_tag_type}', - ao3_tag_category='{row[tag_headers['ao3_tag_category']]}', - ao3_tag_fandom='{fandom}' - WHERE {tagid_filter} - """ + query, + [ + str(ao3_tag), + ao3_tag_type, + row[tag_headers["ao3_tag_category"]], + fandom, + ], ) return num_insert def tags_by_story_id(self, item_type: str = "story"): story_ids = self.sql.execute_and_fetchall( self.database, - f""" + """ SELECT item_id, item_type, GROUP_CONCAT(tag_id) as tag_ids - FROM item_tags WHERE item_type='{item_type}' GROUP BY item_id, item_type ;""", + FROM item_tags WHERE item_type=%s GROUP BY item_id, item_type""", + (item_type), ) - cur = 0 total = len(story_ids) tags_by_story_id = {} - for story_id in story_ids: - cur += 1 + for i, story_id in enumerate(story_ids): sys.stdout.write( - f"\rCollecting tags for {item_type}: {cur}/{total} (including Do Not Import)" + f"\rCollecting tags for {item_type}: {i}/{total} (including Do Not Import)" ) sys.stdout.flush() tags = self.sql.execute_dict( - f"SELECT * FROM tags WHERE id in ({story_id[2]}) AND ao3_tag != ''" + "SELECT * FROM tags WHERE id in %s AND ao3_tag != ''", (story_id[2]) ) tags_by_story_id[story_id[0]] = tags return tags_by_story_id diff --git a/shared_python/__init__.py b/shared_python/__init__.py old mode 100755 new mode 100644 diff --git a/test/test_multiple_authors.py b/test/test_multiple_authors.py index f2337b5..e1596dc 100644 --- a/test/test_multiple_authors.py +++ b/test/test_multiple_authors.py @@ -1,15 +1,16 @@ -from unittest import TestCase -import unittest -from shared_python.Logging import logger -from shared_python.FinalTables import FinalTables import argparse import datetime +import unittest +from unittest import TestCase + +from shared_python.FinalTables import FinalTables +from shared_python.Logging import logger def testArgs(): args = argparse.Namespace() - setattr(args, "temp_db_database", "test_final_open_doors") - setattr(args, "output_database", "test_final_open_doors") + args.temp_db_database = "test_final_open_doors" + args.output_database = "test_final_open_doors" return args diff --git a/test/test_multitag_mapping.py b/test/test_multitag_mapping.py index 51fa29d..e50e51b 100644 --- a/test/test_multitag_mapping.py +++ b/test/test_multitag_mapping.py @@ -1,22 +1,23 @@ -from unittest import TestCase +import argparse import unittest -from shared_python.Sql import Sql +from unittest import TestCase + from shared_python.Logging import logger +from shared_python.Sql import Sql from shared_python.Tags import Tags -import argparse def testArgs(): parser = argparse.ArgumentParser(description="Test an archive database") args = parser.parse_args() - setattr(args, "archive_type", "AA") - setattr(args, "db_host", "localhost") - setattr(args, "db_user", "root") - setattr(args, "db_password", "test") - setattr(args, "temp_db_database", "test_working_open_doors") - setattr(args, "output_database", "unit_test_output") - setattr(args, "default_fandom", "Fandom C (TV)") - setattr(args, "sql_path", "./test/test_data/test.sql") + args.archive_type = "AA" + args.db_host = "localhost" + args.db_user = "root" + args.db_password = "test" + args.temp_db_database = "test_working_open_doors" + args.output_database = "unit_test_output" + args.default_fandom = "Fandom C (TV)" + args.sql_path = "./test/test_data/test.sql" return args diff --git a/test/test_percent_symbol.py b/test/test_percent_symbol.py index ad3d01e..389920c 100644 --- a/test/test_percent_symbol.py +++ b/test/test_percent_symbol.py @@ -1,23 +1,24 @@ -from unittest import TestCase -import unittest -from shared_python.Sql import Sql -from shared_python.Logging import logger -from shared_python.FinalTables import FinalTables import argparse import datetime +import unittest +from unittest import TestCase + +from shared_python.FinalTables import FinalTables +from shared_python.Logging import logger +from shared_python.Sql import Sql def testArgs(): parser = argparse.ArgumentParser(description="Test an archive database") args = parser.parse_args() - setattr(args, "archive_type", "AA") - setattr(args, "db_host", "localhost") - setattr(args, "db_user", "root") - setattr(args, "db_password", "test") - setattr(args, "temp_db_database", "test_final_open_doors") - setattr(args, "output_database", "test_final_open_doors") - setattr(args, "default_fandom", "Fandom C (TV)") - setattr(args, "sql_path", "./test/test_data/test_final_tables.sql") + args.archive_type = "AA" + args.db_host = "localhost" + args.db_user = "root" + args.db_password = "test" + args.temp_db_database = "test_final_open_doors" + args.output_database = "test_final_open_doors" + args.default_fandom = "Fandom C (TV)" + args.sql_path = "./test/test_data/test_final_tables.sql" return args diff --git a/test/test_populate_tags.py b/test/test_populate_tags.py index abcf0f5..1fa7840 100644 --- a/test/test_populate_tags.py +++ b/test/test_populate_tags.py @@ -1,24 +1,23 @@ -from unittest import TestCase +import argparse import unittest +from unittest import TestCase from unittest.mock import MagicMock from shared_python.Logging import logger from shared_python.PopulateTags import PopulateTags -import argparse - from shared_python.Tags import Tags def testArgs(): parser = argparse.ArgumentParser(description="Test an archive database") args = parser.parse_args([]) - setattr(args, "archive_type", "AA") - setattr(args, "db_host", "localhost") - setattr(args, "db_user", "root") - setattr(args, "db_password", "") - setattr(args, "temp_db_database", "unit_test") - setattr(args, "output_database", "unit_test_output") - setattr(args, "default_fandom", "Fandom C (TV)") + args.archive_type = "AA" + args.db_host = "localhost" + args.db_user = "root" + args.db_password = "" + args.temp_db_database = "unit_test" + args.output_database = "unit_test_output" + args.default_fandom = "Fandom C (TV)" return args @@ -30,7 +29,7 @@ class TestPopulate_tags(TestCase): final = MagicMock() populate_tags = PopulateTags(args, sql, log, tags, final) - basic_tags = { + basic_tags = { # noqa: RUF012 "fandoms": [ {"original_tag": "Fandom A", "ao3_tag": "Fandom A (TV)"}, {"original_tag": "Fandom B", "ao3_tag": "Fandom B (TV)"}, diff --git a/test/test_step_05.py b/test/test_step_05.py index bd7a2a9..d576c02 100644 --- a/test/test_step_05.py +++ b/test/test_step_05.py @@ -1,13 +1,13 @@ +import argparse +import importlib +import os +import sys +import unittest from collections import defaultdict from unittest import TestCase -import unittest -from shared_python.Sql import Sql -from shared_python.Logging import logger -import argparse -import sys -import os -import importlib +from shared_python.Logging import logger +from shared_python.Sql import Sql current = os.path.dirname(os.path.realpath(__file__)) parent = os.path.dirname(current) @@ -19,16 +19,16 @@ def testArgs(): parser = argparse.ArgumentParser(description="Test an archive database") args = parser.parse_args() - setattr(args, "archive_type", "AA") - setattr(args, "db_host", "localhost") - setattr(args, "db_user", "root") - setattr(args, "db_password", "test") - setattr(args, "temp_db_database", "test_working_step_five") - setattr(args, "output_database", "unit_test_step_five") - setattr(args, "default_fandom", "Fandom C (TV)") - setattr(args, "sql_path", "./test/test_data/test_working_tables.sql") - setattr(args, "story_ids_to_remove", "") - setattr(args, "bookmark_ids_to_remove", "") + args.archive_type = "AA" + args.db_host = "localhost" + args.db_user = "root" + args.db_password = "test" + args.temp_db_database = "test_working_step_five" + args.output_database = "unit_test_step_five" + args.default_fandom = "Fandom C (TV)" + args.sql_path = "./test/test_data/test_working_tables.sql" + args.story_ids_to_remove = "" + args.bookmark_ids_to_remove = "" return args @@ -41,7 +41,7 @@ class TestStepFive(TestCase): def test_chapter_table_structure(self): step_05.main(self.args, self.log) - get_indexes = "show index from {}.chapters".format(self.args.output_database) + get_indexes = f"show index from {self.args.output_database}.chapters" result = self.sql.execute_dict(get_indexes) col_indexes = defaultdict(list) @@ -51,9 +51,9 @@ def test_chapter_table_structure(self): self.assertEqual(len(col_indexes["story_id"]), 1) def tearDown(self): - drop_working = "drop database {}".format(self.args.temp_db_database) + drop_working = f"drop database {self.args.temp_db_database}" self.sql.execute(drop_working, database=self.args.temp_db_database) - drop_final = "drop database {}".format(self.args.output_database) + drop_final = f"drop database {self.args.output_database}" self.sql.execute(drop_final, database=self.args.output_database) diff --git a/test/test_tags_length.py b/test/test_tags_length.py index a92c7e5..942bde0 100644 --- a/test/test_tags_length.py +++ b/test/test_tags_length.py @@ -1,23 +1,24 @@ -from unittest import TestCase -import unittest -from shared_python.Sql import Sql -from shared_python.Logging import logger -from shared_python.FinalTables import FinalTables import argparse import datetime +import unittest +from unittest import TestCase + +from shared_python.FinalTables import FinalTables +from shared_python.Logging import logger +from shared_python.Sql import Sql def testArgs(): parser = argparse.ArgumentParser(description="Test an archive database") args = parser.parse_args() - setattr(args, "archive_type", "AA") - setattr(args, "db_host", "localhost") - setattr(args, "db_user", "root") - setattr(args, "db_password", "test") - setattr(args, "temp_db_database", "test_final_open_doors") - setattr(args, "output_database", "test_final_open_doors") - setattr(args, "default_fandom", "Fandom C (TV)") - setattr(args, "sql_path", "./shared_python/create-open-doors-tables.sql") + args.archive_type = "AA" + args.db_host = "localhost" + args.db_user = "root" + args.db_password = "test" + args.temp_db_database = "test_final_open_doors" + args.output_database = "test_final_open_doors" + args.default_fandom = "Fandom C (TV)" + args.sql_path = "./shared_python/create-open-doors-tables.sql" return args diff --git a/xx-Remove-DNI-from-Open-Doors-Tables.py b/xx-Remove-DNI-from-Open-Doors-Tables.py index 60eccae..0731703 100755 --- a/xx-Remove-DNI-from-Open-Doors-Tables.py +++ b/xx-Remove-DNI-from-Open-Doors-Tables.py @@ -1,4 +1,5 @@ -# encoding: utf-8 +#!/usr/bin/env python3 + import os from shared_python.Args import Args @@ -19,9 +20,7 @@ for line in f: story_exclusion_filter = filter + "(" + line + ")" - command = "DELETE FROM `{}`.`stories` {}".format( - args.output_database, story_exclusion_filter - ) + command = f"DELETE FROM `{args.output_database}`.`stories` {story_exclusion_filter}" print(command) result = sql.execute(command) print(result) diff --git a/xx-Remove-emails-from-Open-Doors-Tables.py b/xx-Remove-emails-from-Open-Doors-Tables.py index ebd34b3..d5fe889 100755 --- a/xx-Remove-emails-from-Open-Doors-Tables.py +++ b/xx-Remove-emails-from-Open-Doors-Tables.py @@ -1,11 +1,14 @@ -#!/bin/python3 -from shared_python.Args import Args -from shared_python.Sql import Sql +#!/usr/bin/env python3 + import re + from prompt_toolkit import print_formatted_text from prompt_toolkit.formatted_text import FormattedText from prompt_toolkit.shortcuts import clear +from shared_python.Args import Args +from shared_python.Sql import Sql + # This regex is pulled from the HTML5 spec. Though it is technically not # compliant with RFC 5322 ("a willful violation"), it's good enough for our # purposes. @@ -53,7 +56,7 @@ def ask_user_for_action(match) -> str: while True: try: return return_from_list(match) - except: # noqa: E722 + except ListException: response = input( f"\n{raw_email} ([W]hitelist, [B]lacklist) ([A]ddress, [D]omain) [C]ontext [R]ewrite domain >\n\t" ).lower() @@ -78,6 +81,10 @@ def ask_user_for_action(match) -> str: addresses = {} +class ListException(Exception): + pass + + def return_from_list(match) -> str: start, end = match.span() raw_email = match.string[start:end] @@ -94,7 +101,7 @@ def return_from_list(match) -> str: return raw_email elif domain_entry is False: return BAN_TEXT - raise Exception("Failed to resolve") + raise ListException("Failed to resolve") def escape_for_sql(raw: str) -> str: @@ -144,7 +151,7 @@ def replace_func(email): addresses[raw_email] = False try: return return_from_list(email) - except: # noqa: E722 + except ListException: return ask_user_for_action(email) cleared_text = email_regex.sub(replace_func, text)