format_changelog() - Code Metrics - Inspection of "Merge pull request #61 from goanpeca/feature/all-m..." - spyder-ide/loghub - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b4caba...44c104 )

by Gonzalo

created 2017-01-31 18:19 UTC

format_changelog() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	16
Bugs	2	Features	1

Metric	Value
cc	9
c	16
b	2
f	1
dl	0
loc	57
rs	5.2182

How to fix Long Method

# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright (c) The Spyder Development Team
#
# Licensed under the terms of the MIT License
# (See LICENSE.txt for details)
# -----------------------------------------------------------------------------
"""Loghub filter and formatter."""

# yapf: disable

# Standard library imports
from collections import OrderedDict
import codecs
import re
import time

# Third party imports
from jinja2 import Template

# Local imports
from loghub.core.repo import GitHubRepo
from loghub.templates import (CHANGELOG_GROUPS_TEMPLATE_PATH,
                              CHANGELOG_TEMPLATE_PATH,
                              RELEASE_GROUPS_TEMPLATE_PATH,
                              RELEASE_TEMPLATE_PATH)


# yapf: enable


def filter_prs_by_regex(issues, pr_label_regex):
    """Filter prs by issue regex."""
    filtered_prs = []
    pr_pattern = re.compile(pr_label_regex)

    for issue in issues:
        is_pr = bool(issue.get('pull_request'))
        labels = ' '.join(issue.get('loghub_label_names'))

        if is_pr and pr_label_regex:
            pr_valid = bool(pr_pattern.search(labels))
            if pr_valid:
                filtered_prs.append(issue)
        if is_pr and not pr_label_regex:
            filtered_prs.append(issue)

    return filtered_prs


def filter_issues_by_regex(issues, issue_label_regex):
    """Filter issues by issue regex."""
    filtered_issues = []
    issue_pattern = re.compile(issue_label_regex)

    for issue in issues:
        is_pr = bool(issue.get('pull_request'))
        is_issue = not is_pr
        labels = ' '.join(issue.get('loghub_label_names'))

        if is_issue and issue_label_regex:
            issue_valid = bool(issue_pattern.search(labels))
            if issue_valid:
                filtered_issues.append(issue)
        elif is_issue and not issue_label_regex:
            filtered_issues.append(issue)

    return filtered_issues


def filter_issue_label_groups(issues, issue_label_groups):
    """Filter issues by the label groups."""
    grouped_filtered_issues = OrderedDict()
    print(issue_label_groups)
    if issue_label_groups:
        new_filtered_issues = []
        for label_group_dic in issue_label_groups:
            grouped_filtered_issues[label_group_dic['name']] = []

        for issue in issues:
            for label_group_dic in issue_label_groups:
                labels = issue.get('loghub_label_names')
                label = label_group_dic['label']
                name = label_group_dic['name']
                if label in labels:
                    grouped_filtered_issues[name].append(issue)
                    new_filtered_issues.append(issue)

        # Remove any empty groups
        for group, grouped_issues in grouped_filtered_issues.copy().items():
            if not grouped_issues:
                grouped_filtered_issues.pop(group)
    else:
        new_filtered_issues = issues

    return new_filtered_issues, grouped_filtered_issues


def create_changelog(repo=None,
                     username=None,
                     password=None,
                     token=None,
                     milestone=None,
                     since_tag=None,
                     until_tag=None,
                     branch=None,
                     output_format='changelog',
                     issue_label_regex='',
                     pr_label_regex='',
                     template_file=None,
                     issue_label_groups=None,
                     batch=None):
    """Create changelog data for single and batched mode."""
    gh = GitHubRepo(
        username=username,
        password=password,
        token=token,
        repo=repo, )

    all_changelogs = []
    version_tag_prefix = 'v'

    if batch:
        # This will get all the issues, might eat up the api rate limit!
        base_issues = issues = gh.issues(state='closed', branch=branch)
        if batch == 'milestones':
            milestones = [i.get('title') for i in gh.milestones()]
            empty_items = [None] * len(milestones)
            items = zip(*(milestones, empty_items, empty_items))
        elif batch == 'tags':
            tags = [
                i.get('ref', '').replace('refs/tags/', '') for i in gh.tags()
            ]
            since_tags = [None] + tags
            until_tags = tags + [None]
            empty_items = [None] * len(since_tags)
            items = zip(*(empty_items, since_tags, until_tags))
    else:
        base_issues = None
        if milestone:
            items = [(milestone, None, None)]
        else:
            items = [(None, since_tag, until_tag)]

    for (milestone, since_tag, until_tag) in reversed(items):
        version = until_tag or None
        closed_at = None
        since = None
        until = None

        # Set milestone or from tag
        if milestone and not since_tag:
            milestone_data = gh.milestone(milestone)
            closed_at = milestone_data['closed_at']
            version = milestone

            if version.startswith(version_tag_prefix):
                version = version[len(version_tag_prefix):]

        elif not milestone and since_tag:
            since = gh.tag(since_tag)['tagger']['date']
            if until_tag:
                until = gh.tag(until_tag)['tagger']['date']
                closed_at = until

        # This returns issues and pull requests
        issues = gh.issues(
            milestone=milestone,
            state='closed',
            since=since,
            until=until,
            branch=branch,
            base_issues=base_issues, )

        # Filter by regex if available
        filtered_prs = filter_prs_by_regex(issues, pr_label_regex)
        filtered_issues = filter_issues_by_regex(issues, issue_label_regex)

        # If issue label grouping, filter issues
        new_filtered_issues, issue_label_groups = filter_issue_label_groups(
            filtered_issues, issue_label_groups)

        ch = render_changelog(
            repo,
            new_filtered_issues,
            filtered_prs,
            version,
            closed_at=closed_at,
            output_format=output_format,
            template_file=template_file,
            issue_label_groups=issue_label_groups)

        all_changelogs.append(ch)

    changelog = '\n'.join(all_changelogs)
    write_changelog(changelog=changelog)

    return changelog


def render_changelog(repo,
                     issues,
                     prs,
                     version=None,
                     closed_at=None,
                     output_format='changelog',
                     template_file=None,
                     issue_label_groups=None):
    """Render changelog data on a jinja template."""
    # Header
    if not version:
        version = '<RELEASE_VERSION>'

    if closed_at:
        close_date = closed_at.split('T')[0]
    else:
        close_date = time.strftime("%Y/%m/%d")

    # Load template
    if template_file:
        filepath = template_file
    else:
        if issue_label_groups:
            if output_format == 'changelog':
                filepath = CHANGELOG_GROUPS_TEMPLATE_PATH
            else:
                filepath = RELEASE_GROUPS_TEMPLATE_PATH
        else:
            if output_format == 'changelog':
                filepath = CHANGELOG_TEMPLATE_PATH
            else:
                filepath = RELEASE_TEMPLATE_PATH

    with open(filepath) as f:
        data = f.read()

    repo_owner, repo_name = repo.split('/')
    template = Template(data)
    rendered = template.render(
        issues=issues,
        pull_requests=prs,
        version=version,
        close_date=close_date,
        repo_full_name=repo,
        repo_owner=repo_owner,
        repo_name=repo_name,
        issue_label_groups=issue_label_groups, )

    return rendered


def write_changelog(changelog, output_file='CHANGELOG.temp'):
    """Output rendered result to prompt and file."""
    print('#' * 79)
    print(changelog)
    print('#' * 79)

    with codecs.open(output_file, "w", "utf-8") as f:
        f.write(changelog)


1			# -- coding: utf-8 --
2			# -----------------------------------------------------------------------------
3			# Copyright (c) The Spyder Development Team
4			#
5			# Licensed under the terms of the MIT License
6			# (See LICENSE.txt for details)
7			# -----------------------------------------------------------------------------
8			"""Loghub filter and formatter."""
9
10			# yapf: disable
11
12			# Standard library imports
13			from collections import OrderedDict
14			import codecs
15			import re
16			import time
17
18			# Third party imports
19			from jinja2 import Template
20
21			# Local imports
22			from loghub.core.repo import GitHubRepo
23			from loghub.templates import (CHANGELOG_GROUPS_TEMPLATE_PATH,
24			CHANGELOG_TEMPLATE_PATH,
25			RELEASE_GROUPS_TEMPLATE_PATH,
26			RELEASE_TEMPLATE_PATH)
27
28
29			# yapf: enable
30
31
32			def filter_prs_by_regex(issues, pr_label_regex):
33			"""Filter prs by issue regex."""
34			filtered_prs = []
35			pr_pattern = re.compile(pr_label_regex)
36
37			for issue in issues:
38			is_pr = bool(issue.get('pull_request'))
39			labels = ' '.join(issue.get('loghub_label_names'))
40
41			if is_pr and pr_label_regex:
42			pr_valid = bool(pr_pattern.search(labels))
43			if pr_valid:
44			filtered_prs.append(issue)
45			if is_pr and not pr_label_regex:
46			filtered_prs.append(issue)
47
48			return filtered_prs
49
50
51			def filter_issues_by_regex(issues, issue_label_regex):
52			"""Filter issues by issue regex."""
53			filtered_issues = []
54			issue_pattern = re.compile(issue_label_regex)
55
56			for issue in issues:
57			is_pr = bool(issue.get('pull_request'))
58			is_issue = not is_pr
59			labels = ' '.join(issue.get('loghub_label_names'))
60
61			if is_issue and issue_label_regex:
62			issue_valid = bool(issue_pattern.search(labels))
63			if issue_valid:
64			filtered_issues.append(issue)
65			elif is_issue and not issue_label_regex:
66			filtered_issues.append(issue)
67
68			return filtered_issues
69
70
71			def filter_issue_label_groups(issues, issue_label_groups):
72			"""Filter issues by the label groups."""
73			grouped_filtered_issues = OrderedDict()
74			print(issue_label_groups)
75			if issue_label_groups:
76			new_filtered_issues = []
77			for label_group_dic in issue_label_groups:
78			grouped_filtered_issues[label_group_dic['name']] = []
79
80			for issue in issues:
81			for label_group_dic in issue_label_groups:
82			labels = issue.get('loghub_label_names')
83			label = label_group_dic['label']
84			name = label_group_dic['name']
85			if label in labels:
86			grouped_filtered_issues[name].append(issue)
87			new_filtered_issues.append(issue)
88
89			# Remove any empty groups
90			for group, grouped_issues in grouped_filtered_issues.copy().items():
91			if not grouped_issues:
92			grouped_filtered_issues.pop(group)
93			else:
94			new_filtered_issues = issues
95
96			return new_filtered_issues, grouped_filtered_issues
97
98
99			def create_changelog(repo=None,
100			username=None,
101			password=None,
102			token=None,
103			milestone=None,
104			since_tag=None,
105			until_tag=None,
106			branch=None,
107			output_format='changelog',
108			issue_label_regex='',
109			pr_label_regex='',
110			template_file=None,
111			issue_label_groups=None,
112			batch=None):
113			"""Create changelog data for single and batched mode."""
114			gh = GitHubRepo(
115			username=username,
116			password=password,
117			token=token,
118			repo=repo, )
119
120			all_changelogs = []
121			version_tag_prefix = 'v'
122
123			if batch:
124			# This will get all the issues, might eat up the api rate limit!
125			base_issues = issues = gh.issues(state='closed', branch=branch)
126			if batch == 'milestones':
127			milestones = [i.get('title') for i in gh.milestones()]
128			empty_items = [None] * len(milestones)
129			items = zip(*(milestones, empty_items, empty_items))
130			elif batch == 'tags':
131			tags = [
132			i.get('ref', '').replace('refs/tags/', '') for i in gh.tags()
133			]
134			since_tags = [None] + tags
135			until_tags = tags + [None]
136			empty_items = [None] * len(since_tags)
137			items = zip(*(empty_items, since_tags, until_tags))
138			else:
139			base_issues = None
140			if milestone:
141			items = [(milestone, None, None)]
142			else:
143			items = [(None, since_tag, until_tag)]
144
145			for (milestone, since_tag, until_tag) in reversed(items):
146			version = until_tag or None
147			closed_at = None
148			since = None
149			until = None
150
151			# Set milestone or from tag
152			if milestone and not since_tag:
153			milestone_data = gh.milestone(milestone)
154			closed_at = milestone_data['closed_at']
155			version = milestone
156
157			if version.startswith(version_tag_prefix):
158			version = version[len(version_tag_prefix):]
159
160			elif not milestone and since_tag:
161			since = gh.tag(since_tag)['tagger']['date']
162			if until_tag:
163			until = gh.tag(until_tag)['tagger']['date']
164			closed_at = until
165
166			# This returns issues and pull requests
167			issues = gh.issues(
168			milestone=milestone,
169			state='closed',
170			since=since,
171			until=until,
172			branch=branch,
173			base_issues=base_issues, )
174
175			# Filter by regex if available
176			filtered_prs = filter_prs_by_regex(issues, pr_label_regex)
177			filtered_issues = filter_issues_by_regex(issues, issue_label_regex)
178
179			# If issue label grouping, filter issues
180			new_filtered_issues, issue_label_groups = filter_issue_label_groups(
181			filtered_issues, issue_label_groups)
182
183			ch = render_changelog(
184			repo,
185			new_filtered_issues,
186			filtered_prs,
187			version,
188			closed_at=closed_at,
189			output_format=output_format,
190			template_file=template_file,
191			issue_label_groups=issue_label_groups)
192
193			all_changelogs.append(ch)
194
195			changelog = '\n'.join(all_changelogs)
196			write_changelog(changelog=changelog)
197
198			return changelog
199
200
201			def render_changelog(repo,
202			issues,
203			prs,
204			version=None,
205			closed_at=None,
206			output_format='changelog',
207			template_file=None,
208			issue_label_groups=None):
209			"""Render changelog data on a jinja template."""
210			# Header
211			if not version:
212			version = '<RELEASE_VERSION>'
213
214			if closed_at:
215			close_date = closed_at.split('T')[0]
216			else:
217			close_date = time.strftime("%Y/%m/%d")
218
219			# Load template
220			if template_file:
221			filepath = template_file
222			else:
223			if issue_label_groups:
224			if output_format == 'changelog':
225			filepath = CHANGELOG_GROUPS_TEMPLATE_PATH
226			else:
227			filepath = RELEASE_GROUPS_TEMPLATE_PATH
228			else:
229			if output_format == 'changelog':
230			filepath = CHANGELOG_TEMPLATE_PATH
231			else:
232			filepath = RELEASE_TEMPLATE_PATH
233
234			with open(filepath) as f:
235			data = f.read()
236
237			repo_owner, repo_name = repo.split('/')
238			template = Template(data)
239			rendered = template.render(
240			issues=issues,
241			pull_requests=prs,
242			version=version,
243			close_date=close_date,
244			repo_full_name=repo,
245			repo_owner=repo_owner,
246			repo_name=repo_name,
247			issue_label_groups=issue_label_groups, )
248
249			return rendered
250
251
252			def write_changelog(changelog, output_file='CHANGELOG.temp'):
253			"""Output rendered result to prompt and file."""
254			print('#' * 79)
255			print(changelog)
256			print('#' * 79)
257
258			with codecs.open(output_file, "w", "utf-8") as f:
259			f.write(changelog)
260

spyder-ide / loghub

Push — master ( b4caba...44c104 )

format_changelog() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like