sort
This commit is contained in:
aka paul
@@ -0,0 +1,112 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import Set
|
||||
|
||||
INDEX_FILE = "index.json"
|
||||
ARCHIVE_FOLDER = "archive"
|
||||
RESERVED_PATHS = {
|
||||
'.git', '.gitignore', '.github', 'node_modules', '.vscode',
|
||||
'archive', '.idea', '__pycache__', '.DS_Store', '.env', '.gitea'
|
||||
}
|
||||
|
||||
|
||||
def load_index() -> Set[str]:
|
||||
try:
|
||||
with open(INDEX_FILE, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
print(f"Loaded {INDEX_FILE}")
|
||||
|
||||
indexed_folders = set()
|
||||
for key in data.keys():
|
||||
folder = key.split('/')[0]
|
||||
indexed_folders.add(folder)
|
||||
|
||||
print(f"Found {len(indexed_folders)} indexed folders")
|
||||
return indexed_folders
|
||||
except FileNotFoundError:
|
||||
print(f"Error: {INDEX_FILE} not found")
|
||||
raise
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Error parsing {INDEX_FILE}: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def get_all_folders() -> Set[str]:
|
||||
all_folders = set()
|
||||
try:
|
||||
for item in os.listdir('.'):
|
||||
item_path = os.path.join('.', item)
|
||||
if os.path.isdir(item_path) and item not in RESERVED_PATHS:
|
||||
all_folders.add(item)
|
||||
|
||||
print(f"Found {len(all_folders)} total folders in repo")
|
||||
return all_folders
|
||||
except Exception as e:
|
||||
print(f"Error scanning directories: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def create_archive_folder() -> None:
|
||||
if not os.path.exists(ARCHIVE_FOLDER):
|
||||
os.makedirs(ARCHIVE_FOLDER)
|
||||
print(f"Created '{ARCHIVE_FOLDER}' folder")
|
||||
else:
|
||||
print(f"'{ARCHIVE_FOLDER}' folder already exists")
|
||||
|
||||
|
||||
def archive_folders(indexed_folders: Set[str], all_folders: Set[str]) -> None:
|
||||
folders_to_archive = all_folders - indexed_folders
|
||||
|
||||
if not folders_to_archive:
|
||||
print(f"\nAll folders are indexed, nothing to archive")
|
||||
return
|
||||
|
||||
print(f"\nFound {len(folders_to_archive)} folders to archive:")
|
||||
|
||||
for folder in sorted(folders_to_archive):
|
||||
source = os.path.join('.', folder)
|
||||
destination = os.path.join(ARCHIVE_FOLDER, folder)
|
||||
|
||||
try:
|
||||
if os.path.exists(destination):
|
||||
print(f"Skipping '{folder}' (already exists in archive)")
|
||||
continue
|
||||
|
||||
shutil.move(source, destination)
|
||||
print(f"Archived: {folder}")
|
||||
except Exception as e:
|
||||
print(f"Error archiving '{folder}': {e}")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main execution function."""
|
||||
print("=" * 60)
|
||||
print("FOLDER ARCHIVAL SCRIPT")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
try:
|
||||
indexed_folders = load_index()
|
||||
|
||||
all_folders = get_all_folders()
|
||||
|
||||
create_archive_folder()
|
||||
|
||||
archive_folders(indexed_folders, all_folders)
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("ARCHIVAL COMPLETE")
|
||||
print("=" * 60)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\nScript failed: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
exit(main())
|
||||
@@ -0,0 +1,168 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import json
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from typing import Dict, Set, Any
|
||||
from pathlib import Path
|
||||
|
||||
API_URL = "https://library.cufiy.net/api/modules.min.json"
|
||||
INDEX_FILE = "index.json"
|
||||
AUTHOR_TO_SYNC = "50/50"
|
||||
|
||||
|
||||
def fetch_api_data() -> list:
|
||||
try:
|
||||
print(f"Fetching data from {API_URL}")
|
||||
req = urllib.request.Request(
|
||||
API_URL,
|
||||
headers={
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
|
||||
}
|
||||
)
|
||||
with urllib.request.urlopen(req) as response:
|
||||
data = json.loads(response.read().decode('utf-8'))
|
||||
print(f"Successfully fetched {len(data)} modules from API")
|
||||
return data
|
||||
except urllib.error.URLError as e:
|
||||
print(f"Error fetching API data: {e}")
|
||||
raise
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Error parsing API response: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def extract_manifest_path(manifest_url: str) -> str:
|
||||
parts = manifest_url.split('/')
|
||||
try:
|
||||
main_index = parts.index('main')
|
||||
path = '/'.join(parts[main_index + 1:])
|
||||
return path
|
||||
except (ValueError, IndexError):
|
||||
return '/'.join(parts[-2:])
|
||||
|
||||
|
||||
def filter_author_modules(api_data: list, author: str) -> Set[str]:
|
||||
author_paths = set()
|
||||
|
||||
for module in api_data:
|
||||
module_author = module.get('author', {})
|
||||
author_name = module_author.get('name') if isinstance(module_author, dict) else module_author
|
||||
|
||||
if author_name == author:
|
||||
manifest_url = module.get('manifestUrl', '')
|
||||
if manifest_url:
|
||||
path = extract_manifest_path(manifest_url)
|
||||
author_paths.add(path)
|
||||
|
||||
print(f"Found {len(author_paths)} modules from author '{author}' in API")
|
||||
return author_paths
|
||||
|
||||
|
||||
def load_index() -> Dict[str, Any]:
|
||||
try:
|
||||
with open(INDEX_FILE, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
print(f"Loaded {len(data)} modules from {INDEX_FILE}")
|
||||
return data
|
||||
except FileNotFoundError:
|
||||
print(f"✗Error: {INDEX_FILE} not found")
|
||||
raise
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Error parsing {INDEX_FILE}: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def is_manga_module(module_type: str) -> bool:
|
||||
return module_type in ("manga", "mangas")
|
||||
|
||||
|
||||
def update_index(index_data: Dict[str, Any], api_paths: Set[str], author: str) -> Dict[str, Any]:
|
||||
updated_index = {}
|
||||
removed_count = 0
|
||||
kept_count = 0
|
||||
manga_protected = 0
|
||||
|
||||
for path, module_data in index_data.items():
|
||||
module_author = module_data.get('author', {})
|
||||
author_name = module_author.get('name') if isinstance(module_author, dict) else module_author
|
||||
module_type = module_data.get('type', '')
|
||||
should_keep = False
|
||||
reason = ""
|
||||
|
||||
# ALWAYS protect manga modules FIRST (highest priority)
|
||||
if is_manga_module(module_type):
|
||||
should_keep = True
|
||||
reason = f"manga module (protected, type: {module_type})"
|
||||
manga_protected += 1
|
||||
elif author_name != author:
|
||||
should_keep = True
|
||||
reason = f"different author ({author_name})"
|
||||
elif path in api_paths:
|
||||
should_keep = True
|
||||
reason = "exists in API"
|
||||
else:
|
||||
reason = f"not in API (author: {author})"
|
||||
|
||||
if should_keep:
|
||||
updated_index[path] = module_data
|
||||
kept_count += 1
|
||||
else:
|
||||
print(f" - Removing: {path} ({reason})")
|
||||
removed_count += 1
|
||||
|
||||
print(f"\nKept {kept_count} modules")
|
||||
if manga_protected > 0:
|
||||
print(f"Protected {manga_protected} manga modules from removal")
|
||||
if removed_count > 0:
|
||||
print(f"Removed {removed_count} modules")
|
||||
else:
|
||||
print(f"No modules needed to be removed")
|
||||
|
||||
return updated_index
|
||||
|
||||
|
||||
def save_index(index_data: Dict[str, Any]) -> None:
|
||||
try:
|
||||
with open(INDEX_FILE, 'w', encoding='utf-8') as f:
|
||||
json.dump(index_data, f, indent=2, ensure_ascii=False)
|
||||
print(f"\nSuccessfully saved updated {INDEX_FILE}")
|
||||
except Exception as e:
|
||||
print(f"Error saving {INDEX_FILE}: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print("INDEX.JSON UPDATE SCRIPT")
|
||||
print("=" * 60)
|
||||
print(f"Syncing author: {AUTHOR_TO_SYNC}")
|
||||
print(f"Protected types: manga, mangas")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
try:
|
||||
api_data = fetch_api_data()
|
||||
|
||||
api_paths = filter_author_modules(api_data, AUTHOR_TO_SYNC)
|
||||
|
||||
index_data = load_index()
|
||||
|
||||
print(f"\nProcessing modules")
|
||||
updated_index = update_index(index_data, api_paths, AUTHOR_TO_SYNC)
|
||||
|
||||
save_index(updated_index)
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("UPDATE COMPLETE")
|
||||
print("=" * 60)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\nScript failed: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
exit(main())
|
||||
Reference in New Issue
Block a user