Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

#!/usr/bin/python 

# -*- coding: utf-8 -*- 

""" 

Create the monuments tables SQL from monuments_config.json files 

 

Author: Platonides 

""" 

import os 

 

import monuments_config as mconfig 

 

 

def process_country(country_config): 

""" 

Process the country configs to create sql files. 

 

@param countryconfig: country configuration 

""" 

# @todo: standardise these as 'monuments_{country}_({lang})' 

table = country_config.get('table') 

 

try: 

if country_config.get('type') == 'sparql': 

sql = process_wikidata_config(country_config) 

else: 

sql = process_classic_config(country_config) 

except Exception as e: 

raise Exception( 

u'{exception} for countrycode: {country}, lang: {lang}'.format( 

exception=e, country=country_config.get('country'), 

lang=country_config.get('lang'))) 

 

f = open(os.path.join( 

get_sql_dir(), u'create_table_{}.sql'.format(table)), 'w') 

f.write(sql) 

f.close() 

 

 

def process_classic_config(country_config): 

""" 

Process a country configuration for wikitext lists. 

 

@param country_config: country configuration 

@return sql 

""" 

primkey = False 

has_lat_lon = False 

default_type = 'varchar(255)' 

source_primkey = country_config.get('primkey') 

fields_sql = [] 

 

for field in country_config.get('fields'): 

column = field.get('dest') 

if column == '': # An entry with no dest is intended to be skipped 

continue 

 

# primkey in config refers to the destination 

if not primkey and source_primkey == field.get('dest'): 

primkey = column 

 

if column in ['lon', 'lat']: 

has_lat_lon = True 

fields_sql.append( 

u'`{}` double DEFAULT NULL,'.format(column.encode('utf8'))) 

else: 

typ = field.get('type') or default_type 

if typ.startswith('int('): 

if field.get('auto_increment'): 

typ += ' NOT NULL AUTO_INCREMENT' 

else: 

typ += ' NOT NULL DEFAULT 0' 

elif typ.startswith("varchar("): 

if field.get('default'): 

typ += u" NOT NULL DEFAULT '{}'".format( 

field.get('default')) 

else: 

typ += " NOT NULL DEFAULT ''" 

 

fields_sql.append(u'`{}` {},'.format(column.encode('utf8'), typ)) 

 

try: 

primkey = validate_primkey(source_primkey, primkey) 

except Exception: 

raise 

 

extra_keys = '' 

if has_lat_lon: 

extra_keys = ( 

",\n" 

"KEY `latitude` (`lat`),\n" 

"KEY `longitude` (`lon`)") 

 

sql = load_classic_template_sql().format( 

table=country_config['table'], 

rows=b'\n '.join(fields_sql), 

primkey=primkey.encode('utf8'), 

extra_keys=extra_keys) 

 

return sql 

 

 

def validate_primkey(source_primkey, primkey): 

""" 

Validate that the primkey was found or construct it from list. 

 

@param source_primkey: the config data on the primkey 

@param primkey: the primkey, if matched 

@return primkey 

@raises Exception 

""" 

if not primkey: 

if source_primkey and not isinstance(source_primkey, (str, unicode)): 

primkey = u"`,`".join(source_primkey) 

else: 

raise Exception('Primary key not found') 

return primkey 

 

 

def load_classic_template_sql(): 

"""Fetch the SQL template for a wikidata config.""" 

filename = 'classic_table.sql.template' 

with open(os.path.join(get_template_dir(), filename), 'r') as f: 

sql = f.read() 

return sql 

 

 

def process_wikidata_config(country_config): 

""" 

Process a country configuration for wikidata sparql queries. 

 

@param country_config: country configuration 

@return sql 

""" 

sql = load_wikidata_template_sql().format( 

table=country_config['table']) 

return sql 

 

 

def load_wikidata_template_sql(): 

"""Fetch the SQL template for a wikidata config.""" 

filename = 'wikidata_table.sql.template' 

with open(os.path.join(get_template_dir(), filename), 'r') as f: 

sql = f.read() 

return sql 

 

 

def get_sql_dir(): 

"""Fetch the SQL template for a wikidata config.""" 

return os.path.join( 

os.path.dirname(os.path.abspath(__file__)), 'sql') 

 

 

def get_template_dir(): 

"""Fetch the SQL template for a wikidata config.""" 

return os.path.join( 

os.path.dirname(os.path.abspath(__file__)), 'template') 

 

 

def main(): 

for (countrycode, lang), countryconfig in mconfig.countries.iteritems(): 

process_country(countryconfig) 

 

 

if __name__ == "__main__": 

try: 

main() 

finally: 

# pywikibot.stopme() 

pass