Skip to content

Commit

Permalink
Merge pull request #54 from bellingcat/dev
Browse files Browse the repository at this point in the history 
dev -> 3.2.0.0
  • Loading branch information
@rly0nheart
rly0nheart authored Dec 3, 2023
2 parents 10a6a08 + 87c4508 commit dc2d149
Show file tree
Hide file tree
Showing 10 changed files with 371 additions and 221 deletions.
6 changes: 3 additions & 3 deletions Knew Karma/KnewKarma/KnewKarma.vbproj
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,11 @@
<PackageProjectUrl>https://github.com/bellingcat/knewkarma/wiki</PackageProjectUrl>
<PackageReadmeFile>README.md</PackageReadmeFile>
<RepositoryUrl>https://github.com/bellingcat/knewkarma</RepositoryUrl>
<AssemblyVersion>3.1.0.0</AssemblyVersion>
<FileVersion>3.1.0.0</FileVersion>
<AssemblyVersion>3.2.0.0</AssemblyVersion>
<FileVersion>3.2.0.0</FileVersion>
<PackageLicenseFile>LICENSE</PackageLicenseFile>
<PackageRequireLicenseAcceptance>True</PackageRequireLicenseAcceptance>
<Version>3.1.0</Version>
<Version>3.2.0</Version>
<PackageTags>reddit;scraper;reddit-scraper;osint;reddit-data</PackageTags>
<PackageReleaseNotes></PackageReleaseNotes>
<AnalysisLevel>6.0-recommended</AnalysisLevel>
Expand Down
8 changes: 4 additions & 4 deletions Knew Karma/KnewKarmaSetup/KnewKarmaSetup.vdproj
View file
Original file line number Diff line number Diff line change
Expand Up @@ -229,15 +229,15 @@
{
"Name" = "8:Microsoft Visual Studio"
"ProductName" = "8:Knew Karma"
"ProductCode" = "8:{024B1DD3-3E98-4906-A6F4-6E200AD1A66C}"
"PackageCode" = "8:{3765B586-DBC1-4999-B7D6-6CDF9930944A}"
"ProductCode" = "8:{73004786-0578-4532-B92C-A9D8A8B5E8B2}"
"PackageCode" = "8:{9ECB32FA-4D0C-4D29-9723-17958851D59F}"
"UpgradeCode" = "8:{9B03AD0F-0C14-4075-AB75-01CD38A594B4}"
"AspNetVersion" = "8:2.0.50727.0"
"RestartWWWService" = "11:FALSE"
"RemovePreviousVersions" = "11:TRUE"
"DetectNewerInstalledVersion" = "11:TRUE"
"InstallAllUsers" = "11:FALSE"
"ProductVersion" = "8:3.1.0"
"ProductVersion" = "8:3.2.0"
"Manufacturer" = "8:Richard Mwewa"
"ARPHELPTELEPHONE" = "8:"
"ARPHELPLINK" = "8:https://github.com/bellingcat/knewkarma/wiki"
Expand Down Expand Up @@ -777,7 +777,7 @@
{
"{5259A561-127C-4D43-A0A1-72F10C7B3BF8}:_6E15D9F422094BD9809550AF1BA1C161"
{
"SourcePath" = "8:..\\KnewKarma\\obj\\Debug\\net6.0-windows\\apphost.exe"
"SourcePath" = "8:..\\KnewKarma\\obj\\Release\\net6.0-windows\\apphost.exe"
"TargetName" = "8:"
"Tag" = "8:"
"Folder" = "8:_C0F76EDD899B4FFF80C2AC1B5526BC22"
Expand Down
50 changes: 42 additions & 8 deletions knewkarma/_cli.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,15 @@

import argparse
import asyncio
import os
from datetime import datetime

import aiohttp
from rich.pretty import pprint

from ._coreutils import log, save_data, pathfinder
from ._parser import create_parser, version
from ._project import PROGRAM_DIRECTORY
from .api import get_updates
from .base import RedditUser, RedditSub, RedditPosts

Expand All @@ -22,10 +24,14 @@ async def setup_cli(arguments: argparse.Namespace):
:param arguments: Argparse namespace object containing parsed command-line arguments.
"""
# -------------------------------------------------------------------- #

data_timeframe: str = arguments.timeframe
data_sorting: str = arguments.limit
data_limit: int = arguments.limit

# -------------------------------------------------------------------- #

user = RedditUser(
username=arguments.username if hasattr(arguments, "username") else None,
data_timeframe=data_timeframe,
Expand All @@ -44,6 +50,8 @@ async def setup_cli(arguments: argparse.Namespace):
limit=data_limit,
)

# -------------------------------------------------------------------- #

# Mapping of command-line commands to their respective functions
function_mapping: dict = {
"user": [
Expand All @@ -70,6 +78,8 @@ async def setup_cli(arguments: argparse.Namespace):
],
}

# -------------------------------------------------------------------- #

if arguments.mode in function_mapping:
async with aiohttp.ClientSession() as request_session:
await get_updates(session=request_session)
Expand All @@ -84,12 +94,26 @@ async def setup_cli(arguments: argparse.Namespace):
pprint(call_function, expand_all=True)
is_executed = True

pathfinder()
save_data(
data=call_function,
to_json=arguments.json,
to_csv=arguments.csv,
)
# -------------------------------------------------------------------- #

if arguments.csv or arguments.json:
target_directory: str = os.path.join(
PROGRAM_DIRECTORY, f"{arguments.mode}_{action}"
)
pathfinder(
directories=[
os.path.join(target_directory, "csv"),
os.path.join(target_directory, "json"),
]
)
save_data(
data=call_function,
save_to_dir=target_directory,
save_json=arguments.json,
save_csv=arguments.csv,
)

# -------------------------------------------------------------------- #

if not is_executed:
log.warning(
Expand All @@ -102,10 +126,15 @@ async def setup_cli(arguments: argparse.Namespace):

def execute_cli():
"""Main entrypoint for the Knew Karma command-line interface."""

# -------------------------------------------------------------------- #

parser = create_parser()
arguments: argparse = parser.parse_args()

start_time: datetime = datetime.now()

# -------------------------------------------------------------------- #

if arguments.mode:
print(
"""
Expand All @@ -114,18 +143,23 @@ def execute_cli():
┛┗┛┛┗┗ ┗┻┛ ┛┗┛┗┻┛ ┛┗┗┗┻"""
)

# -------------------------------------------------------------------- #

try:
start_time: datetime = datetime.now()

log.info(
f"[bold]Knew Karma CLI[/] {version} started at "
f"{start_time.strftime('%a %b %d %Y, %I:%M:%S %p')}..."
f"{start_time.strftime('%a %b %d %Y, %I:%M:%S%p')}..."
)
asyncio.run(setup_cli(arguments=arguments))
except KeyboardInterrupt:
log.warning(f"User interruption detected ([yellow]Ctrl+C[/])")
finally:
log.info(f"Stopped in {datetime.now() - start_time} seconds.")

# -------------------------------------------------------------------- #

else:
# Display usage information if no mode is provided
parser.print_usage()
Expand Down
134 changes: 68 additions & 66 deletions knewkarma/_coreutils.py
View file
Original file line number Diff line number Diff line change
@@ -1,19 +1,16 @@
# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #

import csv
import json
import logging
import os
from datetime import datetime
from typing import Union, List

from ._parser import create_parser
from .metadata import (
CSV_DIRECTORY,
JSON_DIRECTORY,
)
from .data import Comment, Post, Subreddit, User


# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #


def timestamp_to_utc(timestamp: int) -> str:
Expand All @@ -23,84 +20,89 @@ def timestamp_to_utc(timestamp: int) -> str:
:param timestamp: The UNIX timestamp to be converted.
:return: A formatted datetime.utc string in the format "dd MMMM yyyy, hh:mm:ssAM/PM"
"""
from datetime import datetime

utc_from_timestamp: datetime = datetime.utcfromtimestamp(timestamp)
datetime_string: str = utc_from_timestamp.strftime("%d %B %Y, %I:%M:%S%p")

return datetime_string


# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #


def pathfinder():
def pathfinder(directories: list[str]):
"""
Creates file directories in the user's home directory, if they don't already exist.
Creates directories in knewkarma-data directory of the user's home folder.
:param directories: A list of file directories to create.
"""
directories: list = [
CSV_DIRECTORY,
JSON_DIRECTORY,
]
for directory in directories:
os.makedirs(directory, exist_ok=True)


# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #


def save_data(
data,
to_json: bool = False,
to_csv: bool = False,
data: Union[User, Subreddit, List[Union[Post, Comment]]],
save_to_dir: str,
save_json: bool = False,
save_csv: bool = False,
):
"""
Save the given data to JSON and/or CSV files based on the arguments.
Save the given (Reddit) data to a JSON/CSV file based on the save_csv and save_json parameters.
:param data: The data to be saved, which can be a dict or a list of dicts.
:param to_json: Used to get the True value and the filename for the created JSON file if specified.
:param to_csv: Used to get the True value and the filename for the created CSV file if specified.
:param save_to_dir: Directory to save data to.
:param save_json: Used to get the True value and the filename for the created JSON file if specified.
:param save_csv: Used to get the True value and the filename for the created CSV file if specified.
"""
from .base import User, Subreddit

if to_json or to_csv:
if isinstance(data, (User, Subreddit)):
function_data = data.__dict__
elif isinstance(data, list):
function_data = [item.__dict__ for item in data]
else:
log.error(
f"Got an unexpected data type ({type(data)}), "
f"expected {dict} or {list} of {dict}."
)
return

if to_json:
json_path = os.path.join(JSON_DIRECTORY, f"{to_json}.json")
with open(json_path, "w", encoding="utf-8") as json_file:
json.dump(function_data, json_file, indent=4)
log.info(
f"{os.path.getsize(json_file.name)} bytes written to {json_file.name}"
)

if to_csv:
csv_path = os.path.join(CSV_DIRECTORY, f"{to_csv}.csv")
with open(csv_path, "w", newline="", encoding="utf-8") as csv_file:
writer = csv.writer(csv_file)
if isinstance(function_data, dict):
writer.writerow(function_data.keys())
writer.writerow(function_data.values())
elif isinstance(function_data, list):
if function_data:
writer.writerow(
function_data[0].keys()
) # header from keys of the first item
for item in function_data:
writer.writerow(item.values())
log.info(
f"{os.path.getsize(csv_file.name)} bytes written to {csv_file.name}"
)


# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# -------------------------------------------------------------------- #

if isinstance(data, (User, Subreddit)):
function_data = data.__dict__
elif isinstance(data, list):
function_data = [item.__dict__ for item in data]
else:
log.critical(
f"Got an unexpected data type ({type(data)}), "
f"expected {type(User)}, {type(Subreddit)} or {List[Union[type(Post), type(Comment)]]}."
)
return

# -------------------------------------------------------------------- #

if save_json:
json_path = os.path.join(save_to_dir, "json", f"{save_json}.json")
with open(json_path, "w", encoding="utf-8") as json_file:
json.dump(function_data, json_file, indent=4)
log.info(
f"{os.path.getsize(json_file.name)} bytes written to [link file://{json_file.name}]{json_file.name}"
)

# -------------------------------------------------------------------- #

if save_csv:
csv_path = os.path.join(save_to_dir, "csv", f"{save_csv}.csv")
with open(csv_path, "w", newline="", encoding="utf-8") as csv_file:
writer = csv.writer(csv_file)
if isinstance(function_data, dict):
writer.writerow(function_data.keys())
writer.writerow(function_data.values())
elif isinstance(function_data, list):
if function_data:
writer.writerow(
function_data[0].keys()
) # header from keys of the first item
for item in function_data:
writer.writerow(item.values())
log.info(
f"{os.path.getsize(csv_file.name)} bytes written to [link file://{csv_file.name}]{csv_file.name}"
)


# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #


def setup_logging(debug_mode: bool) -> logging.getLogger:
Expand All @@ -124,8 +126,8 @@ def setup_logging(debug_mode: bool) -> logging.getLogger:
return logging.getLogger("Knew Karma")


# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #

log: logging = setup_logging(debug_mode=create_parser().parse_args().debug)
log: logging.getLogger = setup_logging(debug_mode=create_parser().parse_args().debug)

# +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
Loading

0 comments on commit dc2d149

Please sign in to comment.