kris/tarc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: kris:8b36bf9421abf7fd611d4b33fb73462918af0a23
Branches Tags
kris:main kris:testing kris:aitest
..
compare: kris:6273e833f850a7349fe2368b8bb306df2b8a8532
Branches Tags
kris:main kris:testing kris:aitest

1 Commits
8b36bf9421 ... 6273e833f8

Author SHA1 Message Date
Kris Lamoureux
6273e833f8 testing 2024-10-06 03:23:44 -04:00
5 changed files with 115 additions and 129 deletions
Expand all files Collapse all files

15
Makefile
View File

@@ -1,15 +0,0 @@
.PHONY: default venv install dev clean
default: dev
venv:
@[ ! -d ./venv ] && python3 -m venv venv || true
require: venv
@bash -c "source venv/bin/activate && pip install -r requirements.txt"
dev: require
@bash -c "source venv/bin/activate && pip install -e ."
clean:
rm -rf venv dist tarc.egg-info

5
pyproject.toml
View File

@@ -1,7 +1,4 @@
[project]
name = "tarc"
version = "0.0.1dev3"
version = "0.0.1dev2"
description = "Manage BT archives"
[project.scripts]
tarc = "tarc:main"

20
tarc.sh Executable file
View File

@@ -0,0 +1,20 @@
#!/bin/bash
# Create a virtual environment if it does not exist
if [ ! -d "venv" ]; then
# shellcheck disable=SC1091
python3 -m venv venv && \
source venv/bin/activate && \
pip install -r requirements.txt && \
deactivate
fi
# Activate the virtual environment
# shellcheck disable=SC1091
source venv/bin/activate
# Run the Python script
python tarc/main.py "$@"
# Deactivate the virtual environment
deactivate

5
tarc/__init__.py
View File

@@ -1,5 +0,0 @@
"""
tarc - Manage BT archives
"""
from .main import main

199
tarc/main.py
View File

@@ -154,120 +154,109 @@ def list_clients(conn):
return rows
def main():
"""
Entrypoint of the program.
"""
parser = argparse.ArgumentParser(description="Manage BT archives", prog="tarc")
subparsers = parser.add_subparsers(
dest="command", required=True, help="Available commands"
)
parser = argparse.ArgumentParser(description="Manage BT archives", prog="tarc")
subparsers = parser.add_subparsers(
dest="command", required=True, help="Available commands"
)
scan_parser = subparsers.add_parser("scan", help="Scan command")
scan_parser.add_argument("--debug", action="store_true", help="Enable debug mode")
scan_parser.add_argument(
"--confirm-add", action="store_true", help="Confirm adding a new client"
)
scan_parser.add_argument("-n", "--name", help="Name of client")
scan_parser.add_argument("-d", "--directory", help="Directory to scan")
scan_parser.add_argument("-t", "--type", help="Scan type")
scan_parser.add_argument("-e", "--endpoint", help="Endpoint URL")
scan_parser.add_argument("-u", "--username", help="Username")
scan_parser.add_argument("-p", "--password", help="Password")
scan_parser.add_argument("-s", "--storage", help="Path of sqlite3 database")
scan_parser = subparsers.add_parser("scan", help="Scan command")
scan_parser.add_argument("--debug", action="store_true", help="Enable debug mode")
scan_parser.add_argument(
"--confirm-add", action="store_true", help="Confirm adding a new client"
)
scan_parser.add_argument("-n", "--name", help="Name of client")
scan_parser.add_argument("-d", "--directory", help="Directory to scan")
scan_parser.add_argument("-t", "--type", help="Scan type")
scan_parser.add_argument("-e", "--endpoint", help="Endpoint URL")
scan_parser.add_argument("-u", "--username", help="Username")
scan_parser.add_argument("-p", "--password", help="Password")
scan_parser.add_argument("-s", "--storage", help="Path of sqlite3 database")
args = parser.parse_args()
args = parser.parse_args()
if args.command == "scan":
if args.storage is None:
STORAGE = os.path.expanduser("~/.tarch.db")
if args.command == "scan":
if args.storage is None:
STORAGE = os.path.expanduser("~/.tarch.db")
else:
STORAGE = args.storage
try:
sqlitedb = sqlite3.connect(STORAGE)
tables = list_tables(sqlitedb)
except sqlite3.DatabaseError as e:
print(f'[ERROR]: Database Error "{STORAGE}" ({str(e)})')
sys.exit(1)
if len(tables) == 0:
print(f"[INFO]: Initializing database at {STORAGE}")
init_db(sqlitedb)
cursor = sqlitedb.cursor()
cursor.execute("PRAGMA user_version;")
SCHEMA_FOUND = cursor.fetchone()[0]
cursor.close()
if not SCHEMA == SCHEMA_FOUND:
print(f"[ERROR]: SCHEMA {SCHEMA_FOUND}, expected {SCHEMA}")
sys.exit(1)
if not args.directory is None:
print("[INFO]: --directory is not implemented")
sys.exit(0)
elif not args.endpoint is None:
qb = qbittorrent.Client(args.endpoint)
if qb.qbittorrent_version is None:
print(f'[ERROR]: Couldn\'t find client version at "{args.endpoint}"')
sys.exit(1)
elif not re.match(r"^v?\d+(\.\d+)*$", qb.qbittorrent_version):
print(f'[ERROR]: Invalid version found at "{args.endpoint}"')
if args.debug:
print(f"[DEBUG]: {qb.qbittorrent_version}")
sys.exit(1)
else:
STORAGE = args.storage
try:
sqlitedb = sqlite3.connect(STORAGE)
tables = list_tables(sqlitedb)
except sqlite3.DatabaseError as e:
print(f'[ERROR]: Database Error "{STORAGE}" ({str(e)})')
sys.exit(1)
if len(tables) == 0:
print(f"[INFO]: Initializing database at {STORAGE}")
init_db(sqlitedb)
cursor = sqlitedb.cursor()
cursor.execute("PRAGMA user_version;")
SCHEMA_FOUND = cursor.fetchone()[0]
cursor.close()
if not SCHEMA == SCHEMA_FOUND:
print(f"[ERROR]: SCHEMA {SCHEMA_FOUND}, expected {SCHEMA}")
sys.exit(1)
if not args.directory is None:
print("[INFO]: --directory is not implemented")
sys.exit(0)
elif not args.endpoint is None:
qb = qbittorrent.Client(args.endpoint)
if qb.qbittorrent_version is None:
print(f'[ERROR]: Couldn\'t find client version at "{args.endpoint}"')
sys.exit(1)
elif not re.match(r"^v?\d+(\.\d+)*$", qb.qbittorrent_version):
print(f'[ERROR]: Invalid version found at "{args.endpoint}"')
if args.debug:
print(f"[DEBUG]: {qb.qbittorrent_version}")
sys.exit(1)
else:
print(
f'[INFO]: Found qbittorrent {qb.qbittorrent_version} at "{args.endpoint}"'
)
clients = find_client(sqlitedb, args.endpoint)
if args.confirm_add:
if len(clients) == 0:
if not args.name is None:
now = datetime.now(timezone.utc).isoformat(
sep=" ", timespec="seconds"
)
add_client(sqlitedb, args.name, args.endpoint, now)
print(f"[INFO]: Added client {args.name} ({args.endpoint})")
else:
print("[ERROR]: Must specify --name for a new client")
sys.exit(1)
elif len(clients) == 1:
print(f"[ERROR]: {clients[0][1]} ({clients[0][2]}) already exists")
sys.exit(1)
else:
print(
f"[ERROR]: Multiple clients with the same endpoint: {args.endpoint}"
print(
f'[INFO]: Found qbittorrent {qb.qbittorrent_version} at "{args.endpoint}"'
)
clients = find_client(sqlitedb, args.endpoint)
if args.confirm_add:
if len(clients) == 0:
if not args.name is None:
now = datetime.now(timezone.utc).isoformat(
sep=" ", timespec="seconds"
)
add_client(sqlitedb, args.name, args.endpoint, now)
print(f"[INFO]: Added client {args.name} ({args.endpoint})")
else:
print("[ERROR]: Must specify --name for a new client")
sys.exit(1)
elif len(clients) == 0:
print(f'[ERROR]: Client using endpoint "{args.endpoint}" not found')
print("[ERROR]: Use --confirm-add to add a new endpoint")
sys.exit(1)
elif len(clients) == 1:
torrents = qb.torrents()
print(f"[INFO]: There are {len(torrents)} torrents\n")
for torrent in torrents[:2]:
files = qb.get_torrent_files(torrent["hash"])
trackers = qb.get_torrent_trackers(torrent["hash"])
print(f"[name]: {torrent['name']}")
print(f"[infohash_v1]: {torrent['infohash_v1']}")
print(f"[content_path]: {torrent['content_path']}")
print(f"[magent_uri]: {torrent['magnet_uri'][0:80]}")
print(f"[completed_on]: {torrent['completed']}")
print(f"[trackers]: {len(trackers)}")
print(f"[file_count]: {len(files)}\n")
if args.debug:
print(f"[DEBUG]: {repr(torrent)}")
for elem in trackers:
print(f"[DEBUG]: Tracker {repr(elem)}")
print("\n", end="")
print(f"[ERROR]: {clients[0][1]} ({clients[0][2]}) already exists")
sys.exit(1)
else:
print(
f'[ERROR]: Multiple clients ({len(clients)}) using "{args.endpoint}"'
f"[ERROR]: Multiple clients with the same endpoint: {args.endpoint}"
)
sys.exit(1)
else:
print("[ERROR]: Must specify directory OR client endpoint")
elif len(clients) == 0:
print(f'[ERROR]: Client using endpoint "{args.endpoint}" not found')
print("[ERROR]: Use --confirm-add to add a new endpoint")
sys.exit(1)
if __name__ == "__main__":
main()
elif len(clients) == 1:
torrents = qb.torrents()
print(f"[INFO]: There are {len(torrents)} torrents\n")
for torrent in torrents[:2]:
files = qb.get_torrent_files(torrent["hash"])
trackers = qb.get_torrent_trackers(torrent["hash"])
print(f"[name]: {torrent['name']}")
print(f"[infohash_v1]: {torrent['infohash_v1']}")
print(f"[content_path]: {torrent['content_path']}")
print(f"[magent_uri]: {torrent['magnet_uri'][0:80]}")
print(f"[completed_on]: {torrent['completed']}")
print(f"[trackers]: {len(trackers)}")
print(f"[file_count]: {len(files)}\n")
if args.debug:
print(f"[DEBUG]: {repr(torrent)}")
for elem in trackers:
print(f"[DEBUG]: Tracker {repr(elem)}")
print("\n", end="")
else:
print(f'[ERROR]: Multiple clients ({len(clients)}) using "{args.endpoint}"')
sys.exit(1)
else:
print("[ERROR]: Must specify directory OR client endpoint")
sys.exit(1)