1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

# Copyright 2013-2016 The Distro Tracker Developers 

# See the COPYRIGHT file at the top-level directory of this distribution and 

# at https://deb.li/DTAuthors 

# 

# This file is part of Distro Tracker. It is subject to the license terms 

# in the LICENSE file found in the top-level directory of this 

# distribution and at https://deb.li/DTLicense. No part of Distro Tracker, 

# including this file, may be copied, modified, propagated, or distributed 

# except according to the terms contained in the LICENSE file. 

"""Command to import news from a copy of the old PTS tree.""" 

 

import email 

import os 

from datetime import datetime 

 

from django.core.files.base import ContentFile 

from django.core.management.base import BaseCommand, CommandError 

from django.utils import timezone 

 

from distro_tracker.core.models import EmailNews, News, PackageName 

 

 

class Command(BaseCommand): 

""" 

Import old PTS news. 

 

The imported news' signature information is not automatically extracted. 

""" 

 

def add_arguments(self, parser): 

parser.add_argument('rootdir') 

 

def get_directories(self, root_directory): 

return [ 

os.path.join(root_directory, d) 

for d in os.listdir(root_directory) 

if os.path.isdir(os.path.join(root_directory, d)) 

] 

 

def write(self, message): 

41 ↛ 42line 41 didn't jump to line 42, because the condition on line 41 was never true if self.verbose: 

self.stdout.write(message) 

 

def import_package_news(self, package_directory_name): 

package_name = os.path.basename(package_directory_name) 

self.write('Processing package {pkg}...'.format(pkg=package_name)) 

 

try: 

package = PackageName.objects.get(name=package_name) 

except PackageName.DoesNotExist: 

self.write('Package does not exist. Skipping messages...') 

return 

 

news_directory = os.path.join(package_directory_name, 'news') 

55 ↛ 56line 55 didn't jump to line 56, because the condition on line 55 was never true if not os.path.exists(news_directory): 

self.write("Package has no news directory.") 

return 

 

email_news = [] 

for news_file in sorted(os.listdir(news_directory)): 

news_file_path = os.path.join(news_directory, news_file) 

 

try: 

with open(news_file_path, 'rb') as f: 

65 ↛ 68line 65 didn't jump to line 68, because the condition on line 65 was never false if hasattr(email, 'message_from_binary_file'): 

msg = email.message_from_binary_file(f) 

else: 

msg = email.message_from_file(f) 

69 ↛ 75line 69 didn't jump to line 75, because the condition on line 69 was never false if 'Date' in msg: 

timestamp = email.utils.mktime_tz( 

email.utils.parsedate_tz(msg['Date'])) 

date = datetime.utcfromtimestamp(timestamp) 

date = timezone.make_aware(date, timezone.utc) 

else: 

date = timezone.now() 

 

news_kwargs = EmailNews.get_email_news_parameters(msg) 

content = news_kwargs.pop('file_content') 

news_kwargs['news_file'] = ContentFile(content, 

name='news-file') 

 

email_news.append(News( 

package=package, 

datetime_created=date, 

**news_kwargs)) 

except Exception: 

import traceback 

traceback.print_exc() 

self.write('Problem importing news {}'.format(news_file_path)) 

 

self.write("All news for the package processed. " 

"Bulk creating the instances.") 

News.objects.bulk_create(email_news) 

 

self.write('Complete.') 

 

def import_all_news(self, root_directory): 

for hash_directory in self.get_directories(root_directory): 

for package_directory in self.get_directories(hash_directory): 

self.import_package_news(package_directory) 

 

def handle(self, *args, **kwargs): 

103 ↛ 104line 103 didn't jump to line 104, because the condition on line 103 was never true if 'rootdir' not in kwargs or not kwargs['rootdir']: 

raise CommandError("Root directory of old news not provided") 

self.verbose = int(kwargs.get('verbosity', 1)) > 1 

 

# Hack to be able to set the date created field to something else 

# than now. 

EmailNews._meta.get_field('datetime_created').auto_now_add = False 

 

self.import_all_news(kwargs['rootdir']) 

 

EmailNews._meta.get_field('datetime_created').auto_now_add = True