indymedia
/
epicyon
mirror of https://gitlab.com/bashrc2/epicyon
7
4
Fork 1
Code Issues 14 Releases Wiki Activity

Snake case

main
Bob Mottram 2021-12-30 18:38:36 +00:00
parent fb7208938b
commit ec7384681a
4 changed files with 330 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
briar.py
View File

@ -43,33 +43,33 @@ def get_briar_address(actor_json: {}) -> str:
return '' return ''
def set_briar_address(actor_json: {}, briarAddress: str) -> None: def set_briar_address(actor_json: {}, briar_address: str) -> None:
"""Sets an briar address for the given actor """Sets an briar address for the given actor
""" """
notBriarAddress = False not_briar_address = False
if len(briarAddress) < 50: if len(briar_address) < 50:
notBriarAddress = True not_briar_address = True
if not briarAddress.startswith('briar://'): if not briar_address.startswith('briar://'):
notBriarAddress = True not_briar_address = True
if briarAddress.lower() != briarAddress: if briar_address.lower() != briar_address:
notBriarAddress = True not_briar_address = True
if '"' in briarAddress: if '"' in briar_address:
notBriarAddress = True not_briar_address = True
if ' ' in briarAddress: if ' ' in briar_address:
notBriarAddress = True not_briar_address = True
if '.' in briarAddress: if '.' in briar_address:
notBriarAddress = True not_briar_address = True
if ',' in briarAddress: if ',' in briar_address:
notBriarAddress = True not_briar_address = True
if '<' in briarAddress: if '<' in briar_address:
notBriarAddress = True not_briar_address = True
if not actor_json.get('attachment'): if not actor_json.get('attachment'):
actor_json['attachment'] = [] actor_json['attachment'] = []
# remove any existing value # remove any existing value
propertyFound = None property_found = None
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
if not property_value.get('name'): if not property_value.get('name'):
continue continue
@ -77,11 +77,11 @@ def set_briar_address(actor_json: {}, briarAddress: str) -> None:
continue continue
if not property_value['name'].lower().startswith('briar'): if not property_value['name'].lower().startswith('briar'):
continue continue
propertyFound = property_value property_found = property_value
break break
if propertyFound: if property_found:
actor_json['attachment'].remove(propertyFound) actor_json['attachment'].remove(property_found)
if notBriarAddress: if not_briar_address:
return return
for property_value in actor_json['attachment']: for property_value in actor_json['attachment']:
@ -93,12 +93,12 @@ def set_briar_address(actor_json: {}, briarAddress: str) -> None:
continue continue
if property_value['type'] != 'PropertyValue': if property_value['type'] != 'PropertyValue':
continue continue
property_value['value'] = briarAddress property_value['value'] = briar_address
return return
newBriarAddress = { new_briar_address = {
"name": "Briar", "name": "Briar",
"type": "PropertyValue", "type": "PropertyValue",
"value": briarAddress "value": briar_address
} }
actor_json['attachment'].append(newBriarAddress) actor_json['attachment'].append(new_briar_address)

163
cache.py
View File

@ -17,25 +17,25 @@ from utils import get_file_case_insensitive
from utils import get_user_paths from utils import get_user_paths
def _remove_person_from_cache(base_dir: str, personUrl: str, def _remove_person_from_cache(base_dir: str, person_url: str,
person_cache: {}) -> bool: person_cache: {}) -> bool:
"""Removes an actor from the cache """Removes an actor from the cache
""" """
cacheFilename = base_dir + '/cache/actors/' + \ cache_filename = base_dir + '/cache/actors/' + \
personUrl.replace('/', '#') + '.json' person_url.replace('/', '#') + '.json'
if os.path.isfile(cacheFilename): if os.path.isfile(cache_filename):
try: try:
os.remove(cacheFilename) os.remove(cache_filename)
except OSError: except OSError:
print('EX: unable to delete cached actor ' + str(cacheFilename)) print('EX: unable to delete cached actor ' + str(cache_filename))
if person_cache.get(personUrl): if person_cache.get(person_url):
del person_cache[personUrl] del person_cache[person_url]
def check_for_changed_actor(session, base_dir: str, def check_for_changed_actor(session, base_dir: str,
http_prefix: str, domain_full: str, http_prefix: str, domain_full: str,
personUrl: str, avatarUrl: str, person_cache: {}, person_url: str, avatarUrl: str, person_cache: {},
timeoutSec: int): timeout_sec: int):
"""Checks if the avatar url exists and if not then """Checks if the avatar url exists and if not then
the actor has probably changed without receiving an actor/Person Update. the actor has probably changed without receiving an actor/Person Update.
So clear the actor from the cache and it will be refreshed when the next So clear the actor from the cache and it will be refreshed when the next
@ -45,63 +45,63 @@ def check_for_changed_actor(session, base_dir: str,
return return
if domain_full in avatarUrl: if domain_full in avatarUrl:
return return
if url_exists(session, avatarUrl, timeoutSec, http_prefix, domain_full): if url_exists(session, avatarUrl, timeout_sec, http_prefix, domain_full):
return return
_remove_person_from_cache(base_dir, personUrl, person_cache) _remove_person_from_cache(base_dir, person_url, person_cache)
def store_person_in_cache(base_dir: str, personUrl: str, def store_person_in_cache(base_dir: str, person_url: str,
personJson: {}, person_cache: {}, person_json: {}, person_cache: {},
allowWriteToFile: bool) -> None: allow_write_to_file: bool) -> None:
"""Store an actor in the cache """Store an actor in the cache
""" """
if 'statuses' in personUrl or personUrl.endswith('/actor'): if 'statuses' in person_url or person_url.endswith('/actor'):
# This is not an actor or person account # This is not an actor or person account
return return
curr_time = datetime.datetime.utcnow() curr_time = datetime.datetime.utcnow()
person_cache[personUrl] = { person_cache[person_url] = {
"actor": personJson, "actor": person_json,
"timestamp": curr_time.strftime("%Y-%m-%dT%H:%M:%SZ") "timestamp": curr_time.strftime("%Y-%m-%dT%H:%M:%SZ")
} }
if not base_dir: if not base_dir:
return return
# store to file # store to file
if not allowWriteToFile: if not allow_write_to_file:
return return
if os.path.isdir(base_dir + '/cache/actors'): if os.path.isdir(base_dir + '/cache/actors'):
cacheFilename = base_dir + '/cache/actors/' + \ cache_filename = base_dir + '/cache/actors/' + \
personUrl.replace('/', '#') + '.json' person_url.replace('/', '#') + '.json'
if not os.path.isfile(cacheFilename): if not os.path.isfile(cache_filename):
save_json(personJson, cacheFilename) save_json(person_json, cache_filename)
def get_person_from_cache(base_dir: str, personUrl: str, person_cache: {}, def get_person_from_cache(base_dir: str, person_url: str, person_cache: {},
allowWriteToFile: bool) -> {}: allow_write_to_file: bool) -> {}:
"""Get an actor from the cache """Get an actor from the cache
""" """
# if the actor is not in memory then try to load it from file # if the actor is not in memory then try to load it from file
loadedFromFile = False loaded_from_file = False
if not person_cache.get(personUrl): if not person_cache.get(person_url):
# does the person exist as a cached file? # does the person exist as a cached file?
cacheFilename = base_dir + '/cache/actors/' + \ cache_filename = base_dir + '/cache/actors/' + \
personUrl.replace('/', '#') + '.json' person_url.replace('/', '#') + '.json'
actorFilename = get_file_case_insensitive(cacheFilename) actor_filename = get_file_case_insensitive(cache_filename)
if actorFilename: if actor_filename:
personJson = load_json(actorFilename) person_json = load_json(actor_filename)
if personJson: if person_json:
store_person_in_cache(base_dir, personUrl, personJson, store_person_in_cache(base_dir, person_url, person_json,
person_cache, False) person_cache, False)
loadedFromFile = True loaded_from_file = True
if person_cache.get(personUrl): if person_cache.get(person_url):
if not loadedFromFile: if not loaded_from_file:
# update the timestamp for the last time the actor was retrieved # update the timestamp for the last time the actor was retrieved
curr_time = datetime.datetime.utcnow() curr_time = datetime.datetime.utcnow()
curr_timeStr = curr_time.strftime("%Y-%m-%dT%H:%M:%SZ") curr_time_str = curr_time.strftime("%Y-%m-%dT%H:%M:%SZ")
person_cache[personUrl]['timestamp'] = curr_timeStr person_cache[person_url]['timestamp'] = curr_time_str
return person_cache[personUrl]['actor'] return person_cache[person_url]['actor']
return None return None
@ -110,15 +110,15 @@ def expire_person_cache(person_cache: {}):
""" """
curr_time = datetime.datetime.utcnow() curr_time = datetime.datetime.utcnow()
removals = [] removals = []
for personUrl, cacheJson in person_cache.items(): for person_url, cache_json in person_cache.items():
cacheTime = datetime.datetime.strptime(cacheJson['timestamp'], cache_time = datetime.datetime.strptime(cache_json['timestamp'],
"%Y-%m-%dT%H:%M:%SZ") "%Y-%m-%dT%H:%M:%SZ")
daysSinceCached = (curr_time - cacheTime).days days_since_cached = (curr_time - cache_time).days
if daysSinceCached > 2: if days_since_cached > 2:
removals.append(personUrl) removals.append(person_url)
if len(removals) > 0: if len(removals) > 0:
for personUrl in removals: for person_url in removals:
del person_cache[personUrl] del person_cache[person_url]
print(str(len(removals)) + ' actors were expired from the cache') print(str(len(removals)) + ' actors were expired from the cache')
@ -136,52 +136,55 @@ def get_webfinger_from_cache(handle: str, cached_webfingers: {}) -> {}:
return None return None
def get_person_pub_key(base_dir: str, session, personUrl: str, def get_person_pub_key(base_dir: str, session, person_url: str,
person_cache: {}, debug: bool, person_cache: {}, debug: bool,
project_version: str, http_prefix: str, project_version: str, http_prefix: str,
domain: str, onion_domain: str, domain: str, onion_domain: str,
signing_priv_key_pem: str) -> str: signing_priv_key_pem: str) -> str:
if not personUrl: if not person_url:
return None return None
personUrl = personUrl.replace('#main-key', '') person_url = person_url.replace('#main-key', '')
usersPaths = get_user_paths() users_paths = get_user_paths()
for possibleUsersPath in usersPaths: for possible_users_path in users_paths:
if personUrl.endswith(possibleUsersPath + 'inbox'): if person_url.endswith(possible_users_path + 'inbox'):
if debug: if debug:
print('DEBUG: Obtaining public key for shared inbox') print('DEBUG: Obtaining public key for shared inbox')
personUrl = \ person_url = \
personUrl.replace(possibleUsersPath + 'inbox', '/inbox') person_url.replace(possible_users_path + 'inbox', '/inbox')
break break
personJson = \ person_json = \
get_person_from_cache(base_dir, personUrl, person_cache, True) get_person_from_cache(base_dir, person_url, person_cache, True)
if not personJson: if not person_json:
if debug: if debug:
print('DEBUG: Obtaining public key for ' + personUrl) print('DEBUG: Obtaining public key for ' + person_url)
personDomain = domain person_domain = domain
if onion_domain: if onion_domain:
if '.onion/' in personUrl: if '.onion/' in person_url:
personDomain = onion_domain person_domain = onion_domain
profileStr = 'https://www.w3.org/ns/activitystreams' profile_str = 'https://www.w3.org/ns/activitystreams'
asHeader = { accept_str = \
'Accept': 'application/activity+json; profile="' + profileStr + '"' 'application/activity+json; profile="' + profile_str + '"'
as_header = {
'Accept': accept_str
} }
personJson = \ person_json = \
get_json(signing_priv_key_pem, get_json(signing_priv_key_pem,
session, personUrl, asHeader, None, debug, session, person_url, as_header, None, debug,
project_version, http_prefix, personDomain) project_version, http_prefix, person_domain)
if not personJson: if not person_json:
return None return None
pubKey = None pub_key = None
if personJson.get('publicKey'): if person_json.get('publicKey'):
if personJson['publicKey'].get('publicKeyPem'): if person_json['publicKey'].get('publicKeyPem'):
pubKey = personJson['publicKey']['publicKeyPem'] pub_key = person_json['publicKey']['publicKeyPem']
else: else:
if personJson.get('publicKeyPem'): if person_json.get('publicKeyPem'):
pubKey = personJson['publicKeyPem'] pub_key = person_json['publicKeyPem']
if not pubKey: if not pub_key:
if debug: if debug:
print('DEBUG: Public key not found for ' + personUrl) print('DEBUG: Public key not found for ' + person_url)
store_person_in_cache(base_dir, personUrl, personJson, person_cache, True) store_person_in_cache(base_dir, person_url, person_json,
return pubKey person_cache, True)
return pub_key

172
categories.py
View File

@ -10,27 +10,31 @@ __module_group__ = "RSS Feeds"
import os import os
import datetime import datetime
MAX_TAG_LENGTH = 42
INVALID_HASHTAG_CHARS = (',', ' ', '<', ';', '\\', '"', '&', '#')
def get_hashtag_category(base_dir: str, hashtag: str) -> str: def get_hashtag_category(base_dir: str, hashtag: str) -> str:
"""Returns the category for the hashtag """Returns the category for the hashtag
""" """
categoryFilename = base_dir + '/tags/' + hashtag + '.category' category_filename = base_dir + '/tags/' + hashtag + '.category'
if not os.path.isfile(categoryFilename): if not os.path.isfile(category_filename):
categoryFilename = base_dir + '/tags/' + hashtag.title() + '.category' category_filename = base_dir + '/tags/' + hashtag.title() + '.category'
if not os.path.isfile(categoryFilename): if not os.path.isfile(category_filename):
categoryFilename = \ category_filename = \
base_dir + '/tags/' + hashtag.upper() + '.category' base_dir + '/tags/' + hashtag.upper() + '.category'
if not os.path.isfile(categoryFilename): if not os.path.isfile(category_filename):
return '' return ''
categoryStr = None category_str = None
try: try:
with open(categoryFilename, 'r') as fp: with open(category_filename, 'r') as category_file:
categoryStr = fp.read() category_str = category_file.read()
except OSError: except OSError:
print('EX: unable to read category ' + categoryFilename) print('EX: unable to read category ' + category_filename)
if categoryStr: if category_str:
return categoryStr return category_str
return '' return ''
@ -39,88 +43,87 @@ def get_hashtag_categories(base_dir: str,
category: str = None) -> None: category: str = None) -> None:
"""Returns a dictionary containing hashtag categories """Returns a dictionary containing hashtag categories
""" """
maxTagLength = 42 hashtag_categories = {}
hashtagCategories = {}
if recent: if recent:
curr_time = datetime.datetime.utcnow() curr_time = datetime.datetime.utcnow()
daysSinceEpoch = (curr_time - datetime.datetime(1970, 1, 1)).days days_since_epoch = (curr_time - datetime.datetime(1970, 1, 1)).days
recently = daysSinceEpoch - 1 recently = days_since_epoch - 1
for subdir, dirs, files in os.walk(base_dir + '/tags'): for subdir, dirs, files in os.walk(base_dir + '/tags'):
for f in files: for catfile in files:
if not f.endswith('.category'): if not catfile.endswith('.category'):
continue continue
categoryFilename = os.path.join(base_dir + '/tags', f) category_filename = os.path.join(base_dir + '/tags', catfile)
if not os.path.isfile(categoryFilename): if not os.path.isfile(category_filename):
continue continue
hashtag = f.split('.')[0] hashtag = catfile.split('.')[0]
if len(hashtag) > maxTagLength: if len(hashtag) > MAX_TAG_LENGTH:
continue continue
with open(categoryFilename, 'r') as fp: with open(category_filename, 'r') as fp_category:
categoryStr = fp.read() category_str = fp_category.read()
if not categoryStr: if not category_str:
continue continue
if category: if category:
# only return a dictionary for a specific category # only return a dictionary for a specific category
if categoryStr != category: if category_str != category:
continue continue
if recent: if recent:
tagsFilename = base_dir + '/tags/' + hashtag + '.txt' tags_filename = base_dir + '/tags/' + hashtag + '.txt'
if not os.path.isfile(tagsFilename): if not os.path.isfile(tags_filename):
continue continue
modTimesinceEpoc = \ mod_time_since_epoc = \
os.path.getmtime(tagsFilename) os.path.getmtime(tags_filename)
lastModifiedDate = \ last_modified_date = \
datetime.datetime.fromtimestamp(modTimesinceEpoc) datetime.datetime.fromtimestamp(mod_time_since_epoc)
fileDaysSinceEpoch = \ file_days_since_epoch = \
(lastModifiedDate - (last_modified_date -
datetime.datetime(1970, 1, 1)).days datetime.datetime(1970, 1, 1)).days
if fileDaysSinceEpoch < recently: if file_days_since_epoch < recently:
continue continue
if not hashtagCategories.get(categoryStr): if not hashtag_categories.get(category_str):
hashtagCategories[categoryStr] = [hashtag] hashtag_categories[category_str] = [hashtag]
else: else:
if hashtag not in hashtagCategories[categoryStr]: if hashtag not in hashtag_categories[category_str]:
hashtagCategories[categoryStr].append(hashtag) hashtag_categories[category_str].append(hashtag)
break break
return hashtagCategories return hashtag_categories
def update_hashtag_categories(base_dir: str) -> None: def update_hashtag_categories(base_dir: str) -> None:
"""Regenerates the list of hashtag categories """Regenerates the list of hashtag categories
""" """
categoryListFilename = base_dir + '/accounts/categoryList.txt' category_list_filename = base_dir + '/accounts/categoryList.txt'
hashtagCategories = get_hashtag_categories(base_dir) hashtag_categories = get_hashtag_categories(base_dir)
if not hashtagCategories: if not hashtag_categories:
if os.path.isfile(categoryListFilename): if os.path.isfile(category_list_filename):
try: try:
os.remove(categoryListFilename) os.remove(category_list_filename)
except OSError: except OSError:
print('EX: update_hashtag_categories ' + print('EX: update_hashtag_categories ' +
'unable to delete cached category list ' + 'unable to delete cached category list ' +
categoryListFilename) category_list_filename)
return return
categoryList = [] category_list = []
for categoryStr, hashtagList in hashtagCategories.items(): for category_str, _ in hashtag_categories.items():
categoryList.append(categoryStr) category_list.append(category_str)
categoryList.sort() category_list.sort()
categoryListStr = '' category_list_str = ''
for categoryStr in categoryList: for category_str in category_list:
categoryListStr += categoryStr + '\n' category_list_str += category_str + '\n'
# save a list of available categories for quick lookup # save a list of available categories for quick lookup
try: try:
with open(categoryListFilename, 'w+') as fp: with open(category_list_filename, 'w+') as fp_category:
fp.write(categoryListStr) fp_category.write(category_list_str)
except OSError: except OSError:
print('EX: unable to write category ' + categoryListFilename) print('EX: unable to write category ' + category_list_filename)
def _valid_hashtag_category(category: str) -> bool: def _valid_hashtag_category(category: str) -> bool:
@ -129,9 +132,8 @@ def _valid_hashtag_category(category: str) -> bool:
if not category: if not category:
return False return False
invalidChars = (',', ' ', '<', ';', '\\', '"', '&', '#') for char in INVALID_HASHTAG_CHARS:
for ch in invalidChars: if char in category:
if ch in category:
return False return False
# too long # too long
@ -149,34 +151,34 @@ def set_hashtag_category(base_dir: str, hashtag: str, category: str,
return False return False
if not force: if not force:
hashtagFilename = base_dir + '/tags/' + hashtag + '.txt' hashtag_filename = base_dir + '/tags/' + hashtag + '.txt'
if not os.path.isfile(hashtagFilename): if not os.path.isfile(hashtag_filename):
hashtag = hashtag.title() hashtag = hashtag.title()
hashtagFilename = base_dir + '/tags/' + hashtag + '.txt' hashtag_filename = base_dir + '/tags/' + hashtag + '.txt'
if not os.path.isfile(hashtagFilename): if not os.path.isfile(hashtag_filename):
hashtag = hashtag.upper() hashtag = hashtag.upper()
hashtagFilename = base_dir + '/tags/' + hashtag + '.txt' hashtag_filename = base_dir + '/tags/' + hashtag + '.txt'
if not os.path.isfile(hashtagFilename): if not os.path.isfile(hashtag_filename):
return False return False
if not os.path.isdir(base_dir + '/tags'): if not os.path.isdir(base_dir + '/tags'):
os.mkdir(base_dir + '/tags') os.mkdir(base_dir + '/tags')
categoryFilename = base_dir + '/tags/' + hashtag + '.category' category_filename = base_dir + '/tags/' + hashtag + '.category'
if force: if force:
# don't overwrite any existing categories # don't overwrite any existing categories
if os.path.isfile(categoryFilename): if os.path.isfile(category_filename):
return False return False
categoryWritten = False category_written = False
try: try:
with open(categoryFilename, 'w+') as fp: with open(category_filename, 'w+') as fp_category:
fp.write(category) fp_category.write(category)
categoryWritten = True category_written = True
except OSError as ex: except OSError as ex:
print('EX: unable to write category ' + categoryFilename + print('EX: unable to write category ' + category_filename +
' ' + str(ex)) ' ' + str(ex))
if categoryWritten: if category_written:
if update: if update:
update_hashtag_categories(base_dir) update_hashtag_categories(base_dir)
return True return True
@ -184,18 +186,18 @@ def set_hashtag_category(base_dir: str, hashtag: str, category: str,
return False return False
def guess_hashtag_category(tagName: str, hashtagCategories: {}) -> str: def guess_hashtag_category(tagName: str, hashtag_categories: {}) -> str:
"""Tries to guess a category for the given hashtag. """Tries to guess a category for the given hashtag.
This works by trying to find the longest similar hashtag This works by trying to find the longest similar hashtag
""" """
if len(tagName) < 4: if len(tagName) < 4:
return '' return ''
categoryMatched = '' category_matched = ''
tagMatchedLen = 0 tag_matched_len = 0
for categoryStr, hashtagList in hashtagCategories.items(): for category_str, hashtag_list in hashtag_categories.items():
for hashtag in hashtagList: for hashtag in hashtag_list:
if len(hashtag) < 4: if len(hashtag) < 4:
# avoid matching very small strings which often # avoid matching very small strings which often
# lead to spurious categories # lead to spurious categories
@ -203,13 +205,13 @@ def guess_hashtag_category(tagName: str, hashtagCategories: {}) -> str:
if hashtag not in tagName: if hashtag not in tagName:
if tagName not in hashtag: if tagName not in hashtag:
continue continue
if not categoryMatched: if not category_matched:
tagMatchedLen = len(hashtag) tag_matched_len = len(hashtag)
categoryMatched = categoryStr category_matched = category_str
else: else:
# match the longest tag # match the longest tag
if len(hashtag) > tagMatchedLen: if len(hashtag) > tag_matched_len:
categoryMatched = categoryStr category_matched = category_str
if not categoryMatched: if not category_matched:
return '' return ''
return categoryMatched return category_matched

263
city.py
View File

@ -22,8 +22,10 @@ PERSON_SHOP = 3
PERSON_EVENING = 4 PERSON_EVENING = 4
PERSON_PARTY = 5 PERSON_PARTY = 5
BUSY_STATES = (PERSON_WORK, PERSON_SHOP, PERSON_PLAY, PERSON_PARTY)
def _get_decoy_camera(decoySeed: int) -> (str, str, int):
def _get_decoy_camera(decoy_seed: int) -> (str, str, int):
"""Returns a decoy camera make and model which took the photo """Returns a decoy camera make and model which took the photo
""" """
cameras = [ cameras = [
@ -82,13 +84,13 @@ def _get_decoy_camera(decoySeed: int) -> (str, str, int):
["Google", "Pixel 3"], ["Google", "Pixel 3"],
["Google", "Pixel 3a"] ["Google", "Pixel 3a"]
] ]
randgen = random.Random(decoySeed) randgen = random.Random(decoy_seed)
index = randgen.randint(0, len(cameras) - 1) index = randgen.randint(0, len(cameras) - 1)
serialNumber = randgen.randint(100000000000, 999999999999999999999999) serial_number = randgen.randint(100000000000, 999999999999999999999999)
return cameras[index][0], cameras[index][1], serialNumber return cameras[index][0], cameras[index][1], serial_number
def _get_city_pulse(curr_timeOfDay, decoySeed: int) -> (float, float): def _get_city_pulse(curr_time_of_day, decoy_seed: int) -> (float, float):
"""This simulates expected average patterns of movement in a city. """This simulates expected average patterns of movement in a city.
Jane or Joe average lives and works in the city, commuting in Jane or Joe average lives and works in the city, commuting in
and out of the central district for work. They have a unique and out of the central district for work. They have a unique
@ -97,150 +99,149 @@ def _get_city_pulse(curr_timeOfDay, decoySeed: int) -> (float, float):
Distance from the city centre is in the range 0.0 - 1.0 Distance from the city centre is in the range 0.0 - 1.0
Angle is in radians Angle is in radians
""" """
randgen = random.Random(decoySeed) randgen = random.Random(decoy_seed)
variance = 3 variance = 3
busyStates = (PERSON_WORK, PERSON_SHOP, PERSON_PLAY, PERSON_PARTY) data_decoy_state = PERSON_SLEEP
dataDecoyState = PERSON_SLEEP weekday = curr_time_of_day.weekday()
weekday = curr_timeOfDay.weekday() min_hour = 7 + randint(0, variance)
minHour = 7 + randint(0, variance) max_hour = 17 + randint(0, variance)
maxHour = 17 + randint(0, variance) if curr_time_of_day.hour > min_hour:
if curr_timeOfDay.hour > minHour: if curr_time_of_day.hour <= max_hour:
if curr_timeOfDay.hour <= maxHour:
if weekday < 5: if weekday < 5:
dataDecoyState = PERSON_WORK data_decoy_state = PERSON_WORK
elif weekday == 5: elif weekday == 5:
dataDecoyState = PERSON_SHOP data_decoy_state = PERSON_SHOP
else: else:
dataDecoyState = PERSON_PLAY data_decoy_state = PERSON_PLAY
else: else:
if weekday < 5: if weekday < 5:
dataDecoyState = PERSON_EVENING data_decoy_state = PERSON_EVENING
else: else:
dataDecoyState = PERSON_PARTY data_decoy_state = PERSON_PARTY
randgen2 = random.Random(decoySeed + dataDecoyState) randgen2 = random.Random(decoy_seed + data_decoy_state)
angleRadians = \ angle_radians = \
(randgen2.randint(0, 100000) / 100000) * 2 * math.pi (randgen2.randint(0, 100000) / 100000) * 2 * math.pi
# some people are quite random, others have more predictable habits # some people are quite random, others have more predictable habits
decoyRandomness = randgen.randint(1, 3) decoy_randomness = randgen.randint(1, 3)
# occasionally throw in a wildcard to keep the machine learning guessing # occasionally throw in a wildcard to keep the machine learning guessing
if randint(0, 100) < decoyRandomness: if randint(0, 100) < decoy_randomness:
distanceFromCityCenter = (randint(0, 100000) / 100000) distance_from_city_center = (randint(0, 100000) / 100000)
angleRadians = (randint(0, 100000) / 100000) * 2 * math.pi angle_radians = (randint(0, 100000) / 100000) * 2 * math.pi
else: else:
# what consitutes the central district is fuzzy # what consitutes the central district is fuzzy
centralDistrictFuzz = (randgen.randint(0, 100000) / 100000) * 0.1 central_district_fuzz = (randgen.randint(0, 100000) / 100000) * 0.1
busyRadius = 0.3 + centralDistrictFuzz busy_radius = 0.3 + central_district_fuzz
if dataDecoyState in busyStates: if data_decoy_state in BUSY_STATES:
# if we are busy then we're somewhere in the city center # if we are busy then we're somewhere in the city center
distanceFromCityCenter = \ distance_from_city_center = \
(randgen.randint(0, 100000) / 100000) * busyRadius (randgen.randint(0, 100000) / 100000) * busy_radius
else: else:
# otherwise we're in the burbs # otherwise we're in the burbs
distanceFromCityCenter = busyRadius + \ distance_from_city_center = busy_radius + \
((1.0 - busyRadius) * (randgen.randint(0, 100000) / 100000)) ((1.0 - busy_radius) * (randgen.randint(0, 100000) / 100000))
return distanceFromCityCenter, angleRadians return distance_from_city_center, angle_radians
def parse_nogo_string(nogoLine: str) -> []: def parse_nogo_string(nogo_line: str) -> []:
"""Parses a line from locations_nogo.txt and returns the polygon """Parses a line from locations_nogo.txt and returns the polygon
""" """
nogoLine = nogoLine.replace('\n', '').replace('\r', '') nogo_line = nogo_line.replace('\n', '').replace('\r', '')
polygonStr = nogoLine.split(':', 1)[1] polygon_str = nogo_line.split(':', 1)[1]
if ';' in polygonStr: if ';' in polygon_str:
pts = polygonStr.split(';') pts = polygon_str.split(';')
else: else:
pts = polygonStr.split(',') pts = polygon_str.split(',')
if len(pts) <= 4: if len(pts) <= 4:
return [] return []
polygon = [] polygon = []
for index in range(int(len(pts)/2)): for index in range(int(len(pts)/2)):
if index*2 + 1 >= len(pts): if index*2 + 1 >= len(pts):
break break
longitudeStr = pts[index*2].strip() longitude_str = pts[index*2].strip()
latitudeStr = pts[index*2 + 1].strip() latitude_str = pts[index*2 + 1].strip()
if 'E' in latitudeStr or 'W' in latitudeStr: if 'E' in latitude_str or 'W' in latitude_str:
longitudeStr = pts[index*2 + 1].strip() longitude_str = pts[index*2 + 1].strip()
latitudeStr = pts[index*2].strip() latitude_str = pts[index*2].strip()
if 'E' in longitudeStr: if 'E' in longitude_str:
longitudeStr = \ longitude_str = \
longitudeStr.replace('E', '') longitude_str.replace('E', '')
longitude = float(longitudeStr) longitude = float(longitude_str)
elif 'W' in longitudeStr: elif 'W' in longitude_str:
longitudeStr = \ longitude_str = \
longitudeStr.replace('W', '') longitude_str.replace('W', '')
longitude = -float(longitudeStr) longitude = -float(longitude_str)
else: else:
longitude = float(longitudeStr) longitude = float(longitude_str)
latitude = float(latitudeStr) latitude = float(latitude_str)
polygon.append([latitude, longitude]) polygon.append([latitude, longitude])
return polygon return polygon
def spoof_geolocation(base_dir: str, def spoof_geolocation(base_dir: str,
city: str, curr_time, decoySeed: int, city: str, curr_time, decoy_seed: int,
citiesList: [], cities_list: [],
nogoList: []) -> (float, float, str, str, nogo_list: []) -> (float, float, str, str,
str, str, int): str, str, int):
"""Given a city and the current time spoofs the location """Given a city and the current time spoofs the location
for an image for an image
returns latitude, longitude, N/S, E/W, returns latitude, longitude, N/S, E/W,
camera make, camera model, camera serial number camera make, camera model, camera serial number
""" """
locationsFilename = base_dir + '/custom_locations.txt' locations_filename = base_dir + '/custom_locations.txt'
if not os.path.isfile(locationsFilename): if not os.path.isfile(locations_filename):
locationsFilename = base_dir + '/locations.txt' locations_filename = base_dir + '/locations.txt'
nogoFilename = base_dir + '/custom_locations_nogo.txt' nogo_filename = base_dir + '/custom_locations_nogo.txt'
if not os.path.isfile(nogoFilename): if not os.path.isfile(nogo_filename):
nogoFilename = base_dir + '/locations_nogo.txt' nogo_filename = base_dir + '/locations_nogo.txt'
manCityRadius = 0.1 man_city_radius = 0.1
varianceAtLocation = 0.0004 variance_at_location = 0.0004
default_latitude = 51.8744 default_latitude = 51.8744
default_longitude = 0.368333 default_longitude = 0.368333
default_latdirection = 'N' default_latdirection = 'N'
default_longdirection = 'W' default_longdirection = 'W'
if citiesList: if cities_list:
cities = citiesList cities = cities_list
else: else:
if not os.path.isfile(locationsFilename): if not os.path.isfile(locations_filename):
return (default_latitude, default_longitude, return (default_latitude, default_longitude,
default_latdirection, default_longdirection, default_latdirection, default_longdirection,
"", "", 0) "", "", 0)
cities = [] cities = []
try: try:
with open(locationsFilename, 'r') as f: with open(locations_filename, 'r') as loc_file:
cities = f.readlines() cities = loc_file.readlines()
except OSError: except OSError:
print('EX: unable to read locations ' + locationsFilename) print('EX: unable to read locations ' + locations_filename)
nogo = [] nogo = []
if nogoList: if nogo_list:
nogo = nogoList nogo = nogo_list
else: else:
if os.path.isfile(nogoFilename): if os.path.isfile(nogo_filename):
nogoList = [] nogo_list = []
try: try:
with open(nogoFilename, 'r') as f: with open(nogo_filename, 'r') as nogo_file:
nogoList = f.readlines() nogo_list = nogo_file.readlines()
except OSError: except OSError:
print('EX: unable to read ' + nogoFilename) print('EX: unable to read ' + nogo_filename)
for line in nogoList: for line in nogo_list:
if line.startswith(city + ':'): if line.startswith(city + ':'):
polygon = parse_nogo_string(line) polygon = parse_nogo_string(line)
if polygon: if polygon:
nogo.append(polygon) nogo.append(polygon)
city = city.lower() city = city.lower()
for cityName in cities: for city_name in cities:
if city in cityName.lower(): if city in city_name.lower():
cityFields = cityName.split(':') city_fields = city_name.split(':')
latitude = cityFields[1] latitude = city_fields[1]
longitude = cityFields[2] longitude = city_fields[2]
areaKm2 = 0 area_km2 = 0
if len(cityFields) > 3: if len(city_fields) > 3:
areaKm2 = int(cityFields[3]) area_km2 = int(city_fields[3])
latdirection = 'N' latdirection = 'N'
longdirection = 'E' longdirection = 'E'
if 'S' in latitude: if 'S' in latitude:
@ -252,56 +253,57 @@ def spoof_geolocation(base_dir: str,
latitude = float(latitude) latitude = float(latitude)
longitude = float(longitude) longitude = float(longitude)
# get the time of day at the city # get the time of day at the city
approxTimeZone = int(longitude / 15.0) approx_time_zone = int(longitude / 15.0)
if longdirection == 'E': if longdirection == 'E':
approxTimeZone = -approxTimeZone approx_time_zone = -approx_time_zone
curr_timeAdjusted = curr_time - \ curr_time_adjusted = curr_time - \
datetime.timedelta(hours=approxTimeZone) datetime.timedelta(hours=approx_time_zone)
camMake, camModel, camSerialNumber = \ cam_make, cam_model, cam_serial_number = \
_get_decoy_camera(decoySeed) _get_decoy_camera(decoy_seed)
validCoord = False valid_coord = False
seedOffset = 0 seed_offset = 0
while not validCoord: while not valid_coord:
# patterns of activity change in the city over time # patterns of activity change in the city over time
(distanceFromCityCenter, angleRadians) = \ (distance_from_city_center, angle_radians) = \
_get_city_pulse(curr_timeAdjusted, decoySeed + seedOffset) _get_city_pulse(curr_time_adjusted,
decoy_seed + seed_offset)
# The city radius value is in longitude and the reference # The city radius value is in longitude and the reference
# is Manchester. Adjust for the radius of the chosen city. # is Manchester. Adjust for the radius of the chosen city.
if areaKm2 > 1: if area_km2 > 1:
manRadius = math.sqrt(1276 / math.pi) man_radius = math.sqrt(1276 / math.pi)
radius = math.sqrt(areaKm2 / math.pi) radius = math.sqrt(area_km2 / math.pi)
cityRadiusDeg = (radius / manRadius) * manCityRadius city_radius_deg = (radius / man_radius) * man_city_radius
else: else:
cityRadiusDeg = manCityRadius city_radius_deg = man_city_radius
# Get the position within the city, with some randomness added # Get the position within the city, with some randomness added
latitude += \ latitude += \
distanceFromCityCenter * cityRadiusDeg * \ distance_from_city_center * city_radius_deg * \
math.cos(angleRadians) math.cos(angle_radians)
longitude += \ longitude += \
distanceFromCityCenter * cityRadiusDeg * \ distance_from_city_center * city_radius_deg * \
math.sin(angleRadians) math.sin(angle_radians)
longval = longitude longval = longitude
if longdirection == 'W': if longdirection == 'W':
longval = -longitude longval = -longitude
validCoord = not point_in_nogo(nogo, latitude, longval) valid_coord = not point_in_nogo(nogo, latitude, longval)
if not validCoord: if not valid_coord:
seedOffset += 1 seed_offset += 1
if seedOffset > 100: if seed_offset > 100:
break break
# add a small amount of variance around the location # add a small amount of variance around the location
fraction = randint(0, 100000) / 100000 fraction = randint(0, 100000) / 100000
distanceFromLocation = fraction * fraction * varianceAtLocation distance_from_location = fraction * fraction * variance_at_location
fraction = randint(0, 100000) / 100000 fraction = randint(0, 100000) / 100000
angleFromLocation = fraction * 2 * math.pi angle_from_location = fraction * 2 * math.pi
latitude += distanceFromLocation * math.cos(angleFromLocation) latitude += distance_from_location * math.cos(angle_from_location)
longitude += distanceFromLocation * math.sin(angleFromLocation) longitude += distance_from_location * math.sin(angle_from_location)
# gps locations aren't transcendental, so round to a fixed # gps locations aren't transcendental, so round to a fixed
# number of decimal places # number of decimal places
latitude = int(latitude * 100000) / 100000.0 latitude = int(latitude * 100000) / 100000.0
longitude = int(longitude * 100000) / 100000.0 longitude = int(longitude * 100000) / 100000.0
return (latitude, longitude, latdirection, longdirection, return (latitude, longitude, latdirection, longdirection,
camMake, camModel, camSerialNumber) cam_make, cam_model, cam_serial_number)
return (default_latitude, default_longitude, return (default_latitude, default_longitude,
default_latdirection, default_longdirection, default_latdirection, default_longdirection,
@ -314,33 +316,34 @@ def get_spoofed_city(city: str, base_dir: str,
image metadata image metadata
""" """
city = '' city = ''
cityFilename = acct_dir(base_dir, nickname, domain) + '/city.txt' city_filename = acct_dir(base_dir, nickname, domain) + '/city.txt'
if os.path.isfile(cityFilename): if os.path.isfile(city_filename):
try: try:
with open(cityFilename, 'r') as fp: with open(city_filename, 'r') as city_file:
city = fp.read().replace('\n', '') city = city_file.read().replace('\n', '')
except OSError: except OSError:
print('EX: unable to read ' + cityFilename) print('EX: unable to read ' + city_filename)
return city return city
def _point_in_polygon(poly: [], x: float, y: float) -> bool: def _point_in_polygon(poly: [], x_coord: float, y_coord: float) -> bool:
"""Returns true if the given point is inside the given polygon """Returns true if the given point is inside the given polygon
""" """
n = len(poly) num = len(poly)
inside = False inside = False
p2x = 0.0 p2x = 0.0
p2y = 0.0 p2y = 0.0
xints = 0.0 xints = 0.0
p1x, p1y = poly[0] p1x, p1y = poly[0]
for i in range(n + 1): for i in range(num + 1):
p2x, p2y = poly[i % n] p2x, p2y = poly[i % num]
if y > min(p1y, p2y): if y_coord > min(p1y, p2y):
if y <= max(p1y, p2y): if y_coord <= max(p1y, p2y):
if x <= max(p1x, p2x): if x_coord <= max(p1x, p2x):
if p1y != p2y: if p1y != p2y:
xints = (y - p1y) * (p2x - p1x) / (p2y - p1y) + p1x xints = \
if p1x == p2x or x <= xints: (y_coord - p1y) * (p2x - p1x) / (p2y - p1y) + p1x
if p1x == p2x or x_coord <= xints:
inside = not inside inside = not inside
p1x, p1y = p2x, p2y p1x, p1y = p2x, p2y