__filename__ = "git.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "1.5.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@libreserver.org"
__status__ = "Production"
__module_group__ = "Profile Metadata"
import os
import html
from utils import remove_link_tracking
from utils import acct_dir
from utils import has_object_string_type
from utils import text_in_file
from utils import get_attachment_property_value
from utils import remove_html
from utils import get_attributed_to
def _git_format_content(content: str) -> str:
""" replace html formatting, so that it's more
like the original patch file
"""
patch_str = content.replace('
', '\n').replace('
', '\n')
patch_str = patch_str.replace('
', '').replace('
', '\n')
patch_str = html.unescape(patch_str)
if 'From ' in patch_str:
patch_str = 'From ' + patch_str.split('From ', 1)[1]
return patch_str
def _get_git_project_name(base_dir: str, nickname: str, domain: str,
subject: str) -> str:
"""Returns the project name for a git patch
The project name should be contained within the subject line
and should match against a list of projects which the account
holder wants to receive
"""
git_projects_filename = \
acct_dir(base_dir, nickname, domain) + '/gitprojects.txt'
if not os.path.isfile(git_projects_filename):
return None
subject_line_words = subject.lower().split(' ')
for word in subject_line_words:
if text_in_file(word, git_projects_filename):
return word
return None
def is_git_patch(base_dir: str, nickname: str, domain: str,
message_type: str,
subject: str, content: str,
check_project_name: bool = True) -> bool:
"""Is the given post content a git patch?
"""
if message_type not in ('Note', 'Page', 'Patch'):
return False
# must have a subject line
if not subject:
return False
if '[PATCH]' not in content:
return False
if '---' not in content:
return False
if 'diff ' not in content:
return False
if 'From ' not in content:
return False
if 'From:' not in content:
return False
if 'Date:' not in content:
return False
if 'Subject:' not in content:
return False
if '
' not in content:
if '
' not in content:
return False
if check_project_name:
project_name = \
_get_git_project_name(base_dir, nickname, domain, subject)
if not project_name:
return False
return True
def _get_git_hash(patch_str: str) -> str:
"""Returns the commit hash from a given patch
"""
patch_lines = patch_str.split('\n')
for line in patch_lines:
if line.startswith('From '):
words = line.split(' ')
if len(words) > 1:
if len(words[1]) > 20:
return words[1]
break
return None
def _get_patch_description(patch_str: str) -> str:
"""Returns the description from a given patch
"""
patch_lines = patch_str.split('\n')
description = ''
started = False
for line in patch_lines:
if started:
if line.strip() == '---':
break
description += line + '\n'
if line.startswith('Subject:'):
started = True
return description
def convert_post_to_patch(base_dir: str, nickname: str, domain: str,
post_json_object: {}) -> bool:
"""Detects whether the given post contains a patch
and if so then converts it to a Patch ActivityPub type
"""
if not has_object_string_type(post_json_object, False):
return False
if post_json_object['object']['type'] == 'Patch':
return True
if not post_json_object['object'].get('summary'):
return False
if not post_json_object['object'].get('content'):
return False
if not post_json_object['object'].get('attributedTo'):
return False
if get_attributed_to(post_json_object['object']['attributedTo']) is None:
return False
if not is_git_patch(base_dir, nickname, domain,
post_json_object['object']['type'],
post_json_object['object']['summary'],
post_json_object['object']['content'],
False):
return False
patch_str = _git_format_content(post_json_object['object']['content'])
commit_hash = _get_git_hash(patch_str)
if not commit_hash:
return False
post_json_object['object']['type'] = 'Patch'
# add a commitedBy parameter
if not post_json_object['object'].get('committedBy'):
post_json_object['object']['committedBy'] = \
get_attributed_to(post_json_object['object']['attributedTo'])
post_json_object['object']['hash'] = commit_hash
post_json_object['object']['description'] = {
"mediaType": "text/plain",
"content": _get_patch_description(patch_str)
}
# remove content map
if post_json_object['object'].get('contentMap'):
del post_json_object['object']['contentMap']
print('Converted post to Patch ActivityPub type')
return True
def _git_add_from_handle(patch_str: str, handle: str) -> str:
"""Adds the activitypub handle of the sender to the patch
"""
from_str = 'AP-signed-off-by: '
if from_str in patch_str:
return patch_str
patch_lines = patch_str.split('\n')
patch_str = ''
for line in patch_lines:
patch_str += line + '\n'
if line.startswith('From:'):
if from_str not in patch_str:
patch_str += from_str + handle + '\n'
return patch_str
def receive_git_patch(base_dir: str, nickname: str, domain: str,
message_type: str, subject: str, content: str,
from_nickname: str, from_domain: str) -> bool:
"""Receive a git patch
"""
if not is_git_patch(base_dir, nickname, domain,
message_type, subject, content):
return False
patch_str = _git_format_content(content)
patch_lines = patch_str.split('\n')
patch_filename = None
project_dir = None
patches_dir = acct_dir(base_dir, nickname, domain) + '/patches'
# get the subject line and turn it into a filename
for line in patch_lines:
if line.startswith('Subject:'):
patch_subject = \
line.replace('Subject:', '').replace('/', '|')
patch_subject = patch_subject.replace('[PATCH]', '').strip()
patch_subject = patch_subject.replace(' ', '_')
project_name = \
_get_git_project_name(base_dir, nickname, domain, subject)
if not os.path.isdir(patches_dir):
os.mkdir(patches_dir)
project_dir = patches_dir + '/' + project_name
if not os.path.isdir(project_dir):
os.mkdir(project_dir)
patch_filename = \
project_dir + '/' + patch_subject + '.patch'
break
if not patch_filename:
return False
patch_str = \
_git_add_from_handle(patch_str,
'@' + from_nickname + '@' + from_domain)
try:
with open(patch_filename, 'w+', encoding='utf-8') as fp_patch:
fp_patch.write(patch_str)
patch_notify_filename = \
acct_dir(base_dir, nickname, domain) + '/.newPatchContent'
with open(patch_notify_filename, 'w+',
encoding='utf-8') as fp_patch_notify:
fp_patch_notify.write(patch_str)
return True
except OSError as ex:
print('EX: receive_git_patch ' + patch_filename + ' ' + str(ex))
return False
def get_repo_url(actor_json: {}) -> str:
"""Returns a link used for code repo
"""
if not actor_json.get('attachment'):
return ''
if not isinstance(actor_json['attachment'], list):
return ''
repo_type = ('github', 'gitlab', 'codeberg', 'launchpad',
'sourceforge', 'bitbucket', 'gitea')
for property_value in actor_json['attachment']:
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
continue
if name_value.lower() not in repo_type:
continue
if not property_value.get('type'):
continue
prop_value_name, prop_value = \
get_attachment_property_value(property_value)
if not prop_value:
continue
if not property_value['type'].endswith('PropertyValue'):
continue
if '