cdmlib-io/src/main/java/eu/etaxonomy/cdm/io/stream/CsvStream.java

   1 /**
   2 * Copyright (C) 2009 EDIT
   3 * European Distributed Institute of Taxonomy
   4 * http://www.e-taxonomy.eu
   5 *
   6 * The contents of this file are subject to the Mozilla Public License Version 1.1
   7 * See LICENSE.TXT at the top of this package for the full license terms.
   8 */
   9 package eu.etaxonomy.cdm.io.stream;
  10
  11 import java.io.IOException;
  12 import java.util.HashMap;
  13 import java.util.Map;
  14
  15 import org.apache.logging.log4j.LogManager;import org.apache.logging.log4j.Logger;
  16
  17 import au.com.bytecode.opencsv.CSVReader;
  18 import eu.etaxonomy.cdm.common.CdmUtils;
  19 import eu.etaxonomy.cdm.io.common.IIoObservable;
  20 import eu.etaxonomy.cdm.io.common.ObservableBase;
  21 import eu.etaxonomy.cdm.io.dwca.jaxb.ArchiveEntryBase;
  22 import eu.etaxonomy.cdm.io.dwca.jaxb.Core;
  23 import eu.etaxonomy.cdm.io.dwca.jaxb.Extension;
  24 import eu.etaxonomy.cdm.io.dwca.jaxb.Field;
  25 import eu.etaxonomy.cdm.io.stream.terms.TermUri;
  26
  27 /**
  28  * @author a.mueller
  29  * @since 17.10.2011
  30  *
  31  */
  32 public class CsvStream extends ObservableBase implements IIoObservable,IItemStream{
  33     private static final long serialVersionUID = -4828538731087241715L;
  34
  35     @SuppressWarnings("unused")
  36         private static Logger logger = LogManager.getLogger(CsvStream.class);
  37
  38         private final CSVReader csvReader;
  39         private final ArchiveEntryBase archiveEntry;
  40         private final TermUri term;
  41         private int line = 0;
  42
  43         private StreamItem next;
  44
  45         public CsvStream (CSVReader csvReader, ArchiveEntryBase archiveEntry, int startLine){
  46                 this.csvReader = csvReader;
  47                 this.archiveEntry = archiveEntry;
  48                 String rowType = archiveEntry.getRowType();
  49                 term = TermUri.valueOfUriString(rowType);
  50                 line = startLine;
  51
  52                 //FIXME what if null?
  53         }
  54
  55         @Override
  56         public boolean hasNext(){
  57                 if (next != null){
  58                         return true;
  59                 }else{
  60                         next = readMe();
  61                         return (next != null);
  62                 }
  63         }
  64
  65         @Override
  66         public StreamItem read(){
  67 //              line++;
  68                 return readMe();
  69         }
  70
  71         @Override
  72         public String getItemLocation() {
  73                 return CdmUtils.concat("/", this.getStreamLocation() ,String.valueOf(line));
  74         }
  75
  76         private StreamItem readMe(){
  77                 StreamItem resultItem;
  78                 Map<String, String> resultMap;
  79                 if (next != null){
  80                         resultItem = next;
  81                         next = null;
  82                         return resultItem;
  83                 }else{
  84                         resultMap = new HashMap<>();
  85                         try {
  86                                 String[] next = csvReader.readNext();
  87                                 line++;
  88                                 resultItem = new StreamItem(term, null, getItemLocation());
  89                                 if (next == null){
  90                                         return null;
  91                                 }
  92                                 for (Field field : archiveEntry.getField()){
  93                                 int index = field.getIndex();
  94                                 if (index > next.length -1){
  95                                         int missingFields = archiveEntry.getField().size() -  (next.length -1);
  96                                         String message = "Missing value for archive entry %s %s in line %d of input %s.";
  97                                 String msgMoreLines = missingFields <= 1? "": "and %d more fields" ;
  98                                         msgMoreLines = String.format(msgMoreLines,(missingFields - 1) );
  99                                 message = String.format(message, field.getTerm(), msgMoreLines, line, archiveEntry.getFiles().getLocation());
 100
 101                                 if (countObservers() == 0){
 102                                 throw new RuntimeException(message);
 103                                 } else {
 104                                 StringBuilder messageSB = new StringBuilder(message);
 105                                 messageSB.append(" Line is only partly imported. Csv-Array is [");// + next.toString();
 106
 107                                 for(int i=0;i<next.length;i++) {
 108                                 messageSB.append(next[i]);
 109                                 if(i < next.length-1){
 110                                     messageSB.append(",");
 111                                 } else {
 112                                     messageSB.append("]");
 113                                 }
 114                                 }
 115                                 message = messageSB.toString();
 116                                 fireWarningEvent(message, resultItem.getLocation(), 4);
 117                                 break;
 118                                 }
 119                                 }
 120                                         String value = field.getDefault()!= null? field.getDefault(): next[index];
 121                                         String term = field.getTerm();
 122                                         resultMap.put(term, value);
 123                                 }
 124                                 if (archiveEntry instanceof Core){
 125                                         Core core = (Core)archiveEntry;
 126                                         resultMap.put("id", next[core.getId().getIndex()]);
 127                                 }else if(archiveEntry instanceof Extension){
 128                                         Extension extension = (Extension)archiveEntry;
 129                                         resultMap.put("coreId", next[extension.getCoreid().getIndex()]);
 130                                 }else{
 131                                         throw new RuntimeException("Unhandled archiveEntry type");
 132                                 }
 133
 134
 135                         } catch (IOException e) {
 136                                 //TODO handle as event
 137                                 throw new RuntimeException(e);
 138                         }
 139                         resultItem.map = resultMap;
 140                         if (resultItem.map == null){
 141                                 return null;
 142                         }else {
 143                                 return resultItem;
 144                         }
 145
 146                 }
 147         }
 148
 149         public int getLine(){
 150                 return line;
 151         }
 152
 153
 154         /**
 155          * @return the term
 156          */
 157         @Override
 158         public TermUri getTerm() {
 159                 return term;
 160         }
 161
 162         @Override
 163         public String getStreamLocation() {
 164                 return this.archiveEntry.getFiles().getLocation();
 165         }
 166
 167 //******************************** TO STRING **************************************/
 168
 169         @Override
 170         public String toString(){
 171                 if (archiveEntry == null){
 172                         return super.toString();
 173                 }else{
 174                         return "CsvStream for " + CdmUtils.Nz(archiveEntry.getRowType() + " at line " + line);
 175                 }
 176         }
 177
 178
 179 }